feat: adding Table filterDFS functionaility (#21)

Implementing the same tests as we have in the Go [plugin-sdk](https://github.com/cloudquery/plugin-sdk/blob/main/schema/table_test.go#L58-L276) supporting filtering tables by including and skipping tables based on glob patterns.

Author: mnorbury

Diff for: lib/build.gradle

@@ -34,6 +34,8 @@ dependencies {
     implementation "io.grpc:grpc-testing:1.57.1"
     implementation "io.cloudquery:plugin-pb-java:0.0.5"
     implementation "org.apache.arrow:arrow-vector:12.0.1"
+
+    testImplementation 'org.assertj:assertj-core:3.24.2'
 }
 
 testing {

Diff for: lib/src/main/java/io/cloudquery/helper/GlobMatcher.java

@@ -0,0 +1,23 @@
+package io.cloudquery.helper;
+
+import lombok.Getter;
+
+import java.nio.file.FileSystems;
+import java.nio.file.Path;
+import java.nio.file.PathMatcher;
+
+public class GlobMatcher {
+    private final PathMatcher pathMatcher;
+
+    @Getter
+    private final String stringMatch;
+
+    public GlobMatcher(String stringMatch) {
+        this.stringMatch = stringMatch;
+        this.pathMatcher = FileSystems.getDefault().getPathMatcher("glob:" + stringMatch);
+    }
+
+    public boolean matches(String name) {
+        return pathMatcher.matches(Path.of(name));
+    }
+}

Diff for: lib/src/main/java/io/cloudquery/schema/SchemaException.java

@@ -0,0 +1,22 @@
+package io.cloudquery.schema;
+
+public class SchemaException extends Exception {
+    public SchemaException() {
+    }
+
+    public SchemaException(String message) {
+        super(message);
+    }
+
+    public SchemaException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public SchemaException(Throwable cause) {
+        super(cause);
+    }
+
+    public SchemaException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+        super(message, cause, enableSuppression, writableStackTrace);
+    }
+}

Diff for: lib/src/main/java/io/cloudquery/schema/Table.java

@@ -1,12 +1,16 @@
 package io.cloudquery.schema;
 
+import io.cloudquery.helper.GlobMatcher;
 import lombok.Builder;
 import lombok.Getter;
 
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
+import java.util.function.Predicate;
 
 @Builder(toBuilder = true)
 @Getter
@@ -23,6 +27,67 @@ public static List<Table> flattenTables(List<Table> tables) {
         return flattenMap.values().stream().toList();
     }
 
+    public static List<Table> filterDFS(List<Table> tables, List<String> includeConfiguration, List<String> skipConfiguration, boolean skipDependentTables) throws SchemaException {
+        List<GlobMatcher> includes = includeConfiguration.stream().map(GlobMatcher::new).toList();
+        List<GlobMatcher> excludes = skipConfiguration.stream().map(GlobMatcher::new).toList();
+
+        List<Table> flattenedTables = flattenTables(tables);
+        for (GlobMatcher includeMatcher : includes) {
+            boolean includeMatch = false;
+            for (Table table : flattenedTables) {
+                if (includeMatcher.matches(table.getName())) {
+                    includeMatch = true;
+                    break;
+                }
+            }
+            if (!includeMatch) {
+                throw new SchemaException("table configuration includes a pattern \"" + includeMatcher.getStringMatch() + "\" with no matches");
+            }
+        }
+        for (GlobMatcher excludeMatcher : excludes) {
+            boolean excludeMatch = false;
+            for (Table table : flattenedTables) {
+                if (excludeMatcher.matches(table.getName())) {
+                    excludeMatch = true;
+                    break;
+                }
+            }
+            if (!excludeMatch) {
+                throw new SchemaException("skip configuration includes a pattern \"" + excludeMatcher.getStringMatch() + "\" with no matches");
+            }
+        }
+
+        Predicate<Table> include = table -> {
+            for (GlobMatcher matcher : includes) {
+                if (matcher.matches(table.getName())) {
+                    return true;
+                }
+            }
+            return false;
+        };
+
+        Predicate<Table> exclude = table -> {
+            for (GlobMatcher matcher : excludes) {
+                if (matcher.matches(table.getName())) {
+                    return true;
+                }
+            }
+            return false;
+        };
+
+        return filterDFSFunc(tables, include, exclude, skipDependentTables);
+    }
+
+    private static List<Table> filterDFSFunc(List<Table> tables, Predicate<Table> include, Predicate<Table> exclude, boolean skipDependentTables) {
+        List<Table> filteredTables = new ArrayList<>();
+        for (Table table : tables) {
+            Table filteredTable = table.toBuilder().parent(null).build();
+            Optional<Table> optionalFilteredTable = filteredTable.filterDfs(false, include, exclude, skipDependentTables);
+            optionalFilteredTable.ifPresent(filteredTables::add);
+        }
+        return filteredTables;
+    }
+
     public static int maxDepth(List<Table> tables) {
         int depth = 0;
         if (tables.isEmpty()) {
@@ -39,6 +104,32 @@ public static int maxDepth(List<Table> tables) {
 
     private String name;
 
+    private Table parent;
+
     @Builder.Default
     private List<Table> relations = Collections.emptyList();
+
+    private Optional<Table> filterDfs(boolean parentMatched, Predicate<Table> include, Predicate<Table> exclude, boolean skipDependentTables) {
+        if (exclude.test(this)) {
+            return Optional.empty();
+        }
+        boolean matched = parentMatched && !skipDependentTables;
+        if (include.test(this)) {
+            matched = true;
+        }
+        List<Table> filteredRelations = new ArrayList<>();
+        for (Table relation : relations) {
+            Optional<Table> filteredChild = relation.filterDfs(matched, include, exclude, skipDependentTables);
+            if (filteredChild.isPresent()) {
+                matched = true;
+                filteredRelations.add(filteredChild.get());
+            }
+        }
+        this.relations = filteredRelations;
+        if (matched) {
+            return Optional.of(this);
+        }
+        return Optional.empty();
+    }
+
 }

Diff for: lib/src/test/java/io/cloudquery/helper/GlobMatcherTest.java

@@ -0,0 +1,41 @@
+package io.cloudquery.helper;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class GlobMatcherTest {
+    @Test
+    public void shouldMatchWildcard() {
+        GlobMatcher globMatcher = new GlobMatcher("*");
+
+        assertTrue(globMatcher.matches("aws_ec2_vpc"));
+        assertTrue(globMatcher.matches("aws_ec2_eip"));
+        assertTrue(globMatcher.matches("aws_ec2_instance"));
+    }
+
+    @Test
+    public void shouldMatchWildcardSuffix() {
+        GlobMatcher globMatcher = new GlobMatcher("aws_*");
+
+        assertTrue(globMatcher.matches("aws_ec2_vpc"));
+        assertTrue(globMatcher.matches("aws_ec2_eip"));
+        assertTrue(globMatcher.matches("aws_ec2_instance"));
+
+        assertFalse(globMatcher.matches("gcp_project"));
+        assertFalse(globMatcher.matches("other_aws_resource"));
+    }
+
+    @Test
+    public void shouldMatchWildcardPrefixAndSuffix() {
+        GlobMatcher globMatcher = new GlobMatcher("*ec2*");
+
+        assertTrue(globMatcher.matches("aws_ec2_vpc"));
+        assertTrue(globMatcher.matches("aws_ec2_eip"));
+        assertTrue(globMatcher.matches("aws_ec2_instance"));
+
+        assertFalse(globMatcher.matches("gcp_project"));
+        assertFalse(globMatcher.matches("other_aws_resource"));
+    }
+}