1. use singleton for ClpPlanOptimizerProvider

2. rename splitPath to path in ClpSplit
3. add more properties for equals, hashCode, and toString in ClpTableLayoutHandle
4. remove public from methods in ClpMetadataProvider
5. change most fo log.error to log.warn in ClpMySqlSplitProvider and ClpMySqlMetadataProvider and fix one logging statement in ClpMySqlMetadataProvider
6. fix two minor issues in ClpMetadata
7. throw an exception in getRecordSet
8. improve TestClpOptimizer
9. refactor ClpSchemaTree and fix several bugs

Author: wraymo

presto-clp/pom.xml

@@ -49,12 +49,6 @@
             <artifactId>guice</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>com.google.code.findbugs</groupId>
-            <artifactId>jsr305</artifactId>
-            <optional>true</optional>
-        </dependency>
-
         <dependency>
             <groupId>com.google.guava</groupId>
             <artifactId>guava</artifactId>

presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpColumnHandle.java

@@ -85,7 +85,7 @@ public ColumnMetadata getColumnMetadata()
     @Override
     public int hashCode()
     {
-        return Objects.hash(columnName, columnType);
+        return Objects.hash(columnName, originalColumnName, columnType, nullable);
     }
 
     @Override
@@ -99,14 +99,17 @@ public boolean equals(Object obj)
         }
         ClpColumnHandle other = (ClpColumnHandle) obj;
         return Objects.equals(this.columnName, other.columnName) &&
-                Objects.equals(this.columnType, other.columnType);
+                Objects.equals(this.originalColumnName, other.originalColumnName) &&
+                Objects.equals(this.columnType, other.columnType) &&
+                Objects.equals(this.nullable, other.nullable);
     }
 
     @Override
     public String toString()
     {
         return toStringHelper(this)
                 .add("columnName", columnName)
+                .add("originalColumnName", originalColumnName)
                 .add("columnType", columnType)
                 .add("nullable", nullable)
                 .toString();

presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpConnector.java

@@ -21,8 +21,6 @@
 import com.facebook.presto.spi.connector.ConnectorRecordSetProvider;
 import com.facebook.presto.spi.connector.ConnectorSplitManager;
 import com.facebook.presto.spi.connector.ConnectorTransactionHandle;
-import com.facebook.presto.spi.function.FunctionMetadataManager;
-import com.facebook.presto.spi.function.StandardFunctionResolution;
 import com.facebook.presto.spi.transaction.IsolationLevel;
 
 import javax.inject.Inject;
@@ -36,31 +34,27 @@ public class ClpConnector
 
     private final LifeCycleManager lifeCycleManager;
     private final ClpMetadata metadata;
+    private final ClpPlanOptimizerProvider planOptimizerProvider;
     private final ClpRecordSetProvider recordSetProvider;
     private final ClpSplitManager splitManager;
-    private final FunctionMetadataManager functionManager;
-    private final StandardFunctionResolution functionResolution;
-
     @Inject
     public ClpConnector(LifeCycleManager lifeCycleManager,
                         ClpMetadata metadata,
+                        ClpPlanOptimizerProvider planOptimizerProvider,
                         ClpRecordSetProvider recordSetProvider,
-                        ClpSplitManager splitManager,
-                        FunctionMetadataManager functionManager,
-                        StandardFunctionResolution functionResolution)
+                        ClpSplitManager splitManager)
     {
         this.lifeCycleManager = requireNonNull(lifeCycleManager, "lifeCycleManager is null");
         this.metadata = requireNonNull(metadata, "metadata is null");
+        this.planOptimizerProvider = requireNonNull(planOptimizerProvider, "planOptimizerProvider is null");
         this.recordSetProvider = requireNonNull(recordSetProvider, "recordSetProvider is null");
         this.splitManager = requireNonNull(splitManager, "splitManager is null");
-        this.functionManager = requireNonNull(functionManager, "functionManager is null");
-        this.functionResolution = requireNonNull(functionResolution, "functionResolution is null");
     }
 
     @Override
     public ConnectorPlanOptimizerProvider getConnectorPlanOptimizerProvider()
     {
-        return new ClpPlanOptimizerProvider(functionManager, functionResolution);
+        return planOptimizerProvider;
     }
 
     @Override

presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpMetadata.java

@@ -155,13 +155,13 @@ public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(ConnectorSess
     {
         requireNonNull(prefix, "prefix is null");
         String schemaName = prefix.getSchemaName();
-        if (schemaName != null && !schemaName.equals(DEFAULT_SCHEMA_NAME)) {
+        if (schemaName != null && !listSchemaNames(session).contains(schemaName)) {
             return ImmutableMap.of();
         }
 
         List<SchemaTableName> schemaTableNames;
         if (prefix.getTableName() == null) {
-            schemaTableNames = listTables(session, Optional.of(prefix.getSchemaName()));
+            schemaTableNames = listTables(session, Optional.ofNullable(prefix.getSchemaName()));
         }
         else {
             schemaTableNames = ImmutableList.of(new SchemaTableName(prefix.getSchemaName(), prefix.getTableName()));

presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpModule.java

@@ -32,6 +32,7 @@ protected void setup(Binder binder)
     {
         binder.bind(ClpConnector.class).in(Scopes.SINGLETON);
         binder.bind(ClpMetadata.class).in(Scopes.SINGLETON);
+        binder.bind(ClpPlanOptimizerProvider.class).in(Scopes.SINGLETON);
         binder.bind(ClpRecordSetProvider.class).in(Scopes.SINGLETON);
         binder.bind(ClpSplitManager.class).in(Scopes.SINGLETON);
         configBinder(binder).bindConfig(ClpConfig.class);

presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpRecordSetProvider.java

@@ -31,6 +31,6 @@ public RecordSet getRecordSet(ConnectorTransactionHandle transactionHandle,
                                   ConnectorSplit split,
                                   List<? extends ColumnHandle> columns)
     {
-        return null;
+        throw new UnsupportedOperationException("getRecordSet is not supported");
     }
 }

presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpSplit.java

@@ -21,27 +21,25 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.collect.ImmutableList;
 
-import javax.annotation.Nullable;
-
 import java.util.List;
 
 import static com.facebook.presto.spi.schedule.NodeSelectionStrategy.NO_PREFERENCE;
 
 public class ClpSplit
         implements ConnectorSplit
 {
-    private final String splitPath;
+    private final String path;
 
     @JsonCreator
-    public ClpSplit(@JsonProperty("splitPath") @Nullable String splitPath)
+    public ClpSplit(@JsonProperty("path") String path)
     {
-        this.splitPath = splitPath;
+        this.path = path;
     }
 
     @JsonProperty
-    public String getSplitPath()
+    public String getPath()
     {
-        return splitPath;
+        return path;
     }
 
     @Override

presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpTableLayoutHandle.java

@@ -20,6 +20,8 @@
 import java.util.Objects;
 import java.util.Optional;
 
+import static com.google.common.base.MoreObjects.toStringHelper;
+
 public class ClpTableLayoutHandle
         implements ConnectorTableLayoutHandle
 {
@@ -28,7 +30,7 @@ public class ClpTableLayoutHandle
 
     @JsonCreator
     public ClpTableLayoutHandle(@JsonProperty("table") ClpTableHandle table,
-                                @JsonProperty("query") Optional<String> kqlQuery)
+                                @JsonProperty("kqlQuery") Optional<String> kqlQuery)
     {
         this.table = table;
         this.kqlQuery = kqlQuery;
@@ -56,18 +58,22 @@ public boolean equals(Object o)
             return false;
         }
         ClpTableLayoutHandle that = (ClpTableLayoutHandle) o;
-        return Objects.equals(table, that.table);
+        return Objects.equals(table, that.table) &&
+                Objects.equals(kqlQuery, that.kqlQuery);
     }
 
     @Override
     public int hashCode()
     {
-        return Objects.hash(table);
+        return Objects.hash(table, kqlQuery);
     }
 
     @Override
     public String toString()
     {
-        return table.toString();
+        return toStringHelper(this)
+                .add("table", table)
+                .add("kqlQuery", kqlQuery)
+                .toString();
     }
 }

presto-clp/src/main/java/com/facebook/presto/plugin/clp/metadata/ClpMetadataProvider.java

@@ -23,10 +23,10 @@ public interface ClpMetadataProvider
     /**
      * Returns the list of column handles for the given table.
      */
-    public List<ClpColumnHandle> listColumnHandles(SchemaTableName schemaTableName);
+    List<ClpColumnHandle> listColumnHandles(SchemaTableName schemaTableName);
 
     /**
      * Returns the list of table names in the given schema.
      */
-    public List<String> listTableNames(String schema);
+    List<String> listTableNames(String schema);
 }

presto-clp/src/main/java/com/facebook/presto/plugin/clp/metadata/ClpMySqlMetadataProvider.java

@@ -83,7 +83,7 @@ public List<ClpColumnHandle> listColumnHandles(SchemaTableName schemaTableName)
             }
         }
         catch (SQLException e) {
-            log.error("Failed to load table schema for %s: %s" + schemaTableName.getTableName(), e);
+            log.warn("Failed to load table schema for %s: %s", schemaTableName.getTableName(), e);
         }
         return schemaTree.collectColumnHandles();
     }
@@ -108,7 +108,7 @@ public List<String> listTableNames(String schema)
             }
         }
         catch (SQLException e) {
-            log.error("Failed to load table names: %s", e);
+            log.warn("Failed to load table names: %s", e);
         }
         return tableNames;
     }

presto-clp/src/main/java/com/facebook/presto/plugin/clp/metadata/ClpSchemaTree.java

@@ -41,6 +41,17 @@ static class ClpNode
         Map<String, ClpNode> children = new HashMap<>();
         Set<String> conflictingBaseNames = new HashSet<>();
 
+        ClpNode(String originalName)
+        {
+            this.originalName = originalName;
+        }
+
+        ClpNode(String originalName, Type type)
+        {
+            this.originalName = originalName;
+            this.type = type;
+        }
+
         boolean isLeaf()
         {
             return children.isEmpty();
@@ -52,7 +63,7 @@ boolean isLeaf()
     ClpSchemaTree(boolean polymorphicTypeEnabled)
     {
         this.polymorphicTypeEnabled = polymorphicTypeEnabled;
-        this.root = new ClpNode();
+        this.root = new ClpNode(""); // Root doesn't have an original name
     }
 
     private Type mapColumnType(byte type)
@@ -92,45 +103,23 @@ public void addColumn(String fullName, byte type)
 
         for (int i = 0; i < path.length - 1; i++) {
             String segment = path[i];
-            current.children.putIfAbsent(segment, new ClpNode());
-            current = current.children.get(segment);
+            ClpNode existingNode = current.children.get(segment);
+
+            if (polymorphicTypeEnabled && existingNode != null && existingNode.type != null) {
+                // Conflict: An intermediate segment already exists as a leaf node. Rename it.
+                String existingSuffix = getTypeSuffix(existingNode.type);
+                String renamedExisting = segment + "_" + existingSuffix;
+                current.children.remove(segment);
+                current.children.put(renamedExisting, existingNode);
+            }
+            current = current.children.computeIfAbsent(segment, ClpNode::new);
+            current.type = null;
         }
 
         String leafName = path[path.length - 1];
-        String finalLeafName = leafName;
-
-        if (polymorphicTypeEnabled) {
-            boolean conflictDetected = false;
-
-            if (current.children.containsKey(leafName)) {
-                ClpNode existing = current.children.get(leafName);
-
-                if (existing.type != null && !existing.type.equals(prestoType)) {
-                    String existingSuffix = (existing.type instanceof ArrayType)
-                            ? "array"
-                            : existing.type.getDisplayName();
-                    String renamedExisting = leafName + "_" + existingSuffix;
-
-                    current.children.remove(leafName);
-                    current.children.put(renamedExisting, existing);
-
-                    current.conflictingBaseNames.add(leafName);
-                    conflictDetected = true;
-                }
-            }
-            else if (current.conflictingBaseNames.contains(leafName)) {
-                conflictDetected = true;
-            }
-
-            if (conflictDetected) {
-                String newSuffix = prestoType.getDisplayName();
-                finalLeafName = leafName + "_" + newSuffix;
-            }
-        }
+        String finalLeafName = resolvePolymorphicConflicts(current, leafName, prestoType);
 
-        ClpNode leaf = new ClpNode();
-        leaf.type = prestoType;
-        leaf.originalName = leafName;
+        ClpNode leaf = new ClpNode(leafName, prestoType);
         current.children.put(finalLeafName, leaf);
     }
 
@@ -158,6 +147,44 @@ public List<ClpColumnHandle> collectColumnHandles()
         return columns;
     }
 
+    private String resolvePolymorphicConflicts(ClpNode parent, String baseName, Type newType)
+    {
+        if (!polymorphicTypeEnabled) {
+            return baseName;
+        }
+
+        boolean conflictDetected = false;
+        if (parent.children.containsKey(baseName)) {
+            ClpNode existing = parent.children.get(baseName);
+            if (existing.type == null) {
+                conflictDetected = true;
+            }
+            else if (!existing.type.equals(newType)) {
+                String existingSuffix = getTypeSuffix(existing.type);
+                String renamedExisting = baseName + "_" + existingSuffix;
+                parent.children.remove(baseName);
+                parent.children.put(renamedExisting, existing);
+                parent.conflictingBaseNames.add(baseName);
+                conflictDetected = true;
+            }
+        }
+        else if (parent.conflictingBaseNames.contains(baseName)) {
+            conflictDetected = true;
+        }
+
+        if (conflictDetected) {
+            String newSuffix = getTypeSuffix(newType);
+            return baseName + "_" + newSuffix;
+        }
+
+        return baseName;
+    }
+
+    private String getTypeSuffix(Type type)
+    {
+        return (type instanceof ArrayType) ? "array" : type.getDisplayName();
+    }
+
     private Type buildRowType(ClpNode node)
     {
         List<RowType.Field> fields = new ArrayList<>();

presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/ClpMySqlSplitProvider.java

@@ -81,14 +81,14 @@ public List<ClpSplit> listSplits(ClpTableLayoutHandle clpTableLayoutHandle)
             try (PreparedStatement statement = connection.prepareStatement(tablePathQuery);
                     ResultSet resultSet = statement.executeQuery()) {
                 if (!resultSet.next()) {
-                    log.error("Table metadata not found for table: %s", tableName);
+                    log.warn("Table metadata not found for table: %s", tableName);
                     return ImmutableList.of();
                 }
                 tablePath = resultSet.getString("archive_storage_directory");
             }
 
             if (tablePath == null || tablePath.isEmpty()) {
-                log.error("Table path is null for table: %s", tableName);
+                log.warn("Table path is null for table: %s", tableName);
                 return ImmutableList.of();
             }
 
@@ -103,7 +103,7 @@ public List<ClpSplit> listSplits(ClpTableLayoutHandle clpTableLayoutHandle)
             }
         }
         catch (SQLException e) {
-            log.error("Database error while processing splits for %s: %s", tableName, e);
+            log.warn("Database error while processing splits for %s: %s", tableName, e);
         }
 
         return splits;

presto-clp/src/test/java/com/facebook/presto/plugin/clp/TestClpPlanOptimizer.java

@@ -41,6 +41,9 @@ private void testFilter(String sqlExpression, Optional<String> expectedKqlExpres
             assertTrue(kqlExpression.isPresent());
             assertEquals(kqlExpression.get(), expectedKqlExpression.get());
         }
+        else {
+            assertFalse(kqlExpression.isPresent());
+        }
 
         if (expectedRemainingExpression.isPresent()) {
             assertTrue(remainingExpression.isPresent());