Merge branch 'release-0.293-clp-connector' into xwei/add-k8s-manual-doc

Author: anlowee

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

@@ -24,6 +24,7 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 import static com.facebook.presto.spi.schedule.NodeSelectionStrategy.NO_PREFERENCE;
 import static java.util.Objects.requireNonNull;
@@ -32,11 +33,15 @@ public class ClpSplit
         implements ConnectorSplit
 {
     private final String path;
+    private final Optional<String> kqlQuery;
 
     @JsonCreator
-    public ClpSplit(@JsonProperty("path") String path)
+    public ClpSplit(
+            @JsonProperty("path") String path,
+            @JsonProperty("kqlQuery") Optional<String> kqlQuery)
     {
         this.path = requireNonNull(path, "Split path is null");
+        this.kqlQuery = kqlQuery;
     }
 
     @JsonProperty
@@ -45,6 +50,12 @@ public String getPath()
         return path;
     }
 
+    @JsonProperty
+    public Optional<String> getKqlQuery()
+    {
+        return kqlQuery;
+    }
+
     @Override
     public NodeSelectionStrategy getNodeSelectionStrategy()
     {
@@ -60,6 +71,6 @@ public List<HostAddress> getPreferredNodes(NodeProvider nodeProvider)
     @Override
     public Map<String, String> getInfo()
     {
-        return ImmutableMap.of("path", path);
+        return ImmutableMap.of("path", path, "kqlQuery", kqlQuery.orElse("<null>"));
     }
 }

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

@@ -81,7 +81,7 @@ public List<ClpSplit> listSplits(ClpTableLayoutHandle clpTableLayoutHandle)
                 while (resultSet.next()) {
                     final String archiveId = resultSet.getString(ARCHIVES_TABLE_COLUMN_ID);
                     final String archivePath = tablePath + "/" + archiveId;
-                    splits.add(new ClpSplit(archivePath));
+                    splits.add(new ClpSplit(archivePath, clpTableLayoutHandle.getKqlQuery()));
                 }
             }
         }

presto-native-execution/presto_cpp/main/connectors/PrestoToVeloxConnector.cpp

@@ -1564,7 +1564,7 @@ ClpPrestoToVeloxConnector::toVeloxSplit(
   VELOX_CHECK_NOT_NULL(
       clpSplit, "Unexpected split type {}", connectorSplit->_type);
   return std::make_unique<connector::clp::ClpConnectorSplit>(
-      catalogId, clpSplit->path);
+      catalogId, clpSplit->path, clpSplit->kqlQuery);
 }
 
 std::unique_ptr<velox::connector::ColumnHandle>
@@ -1597,9 +1597,7 @@ ClpPrestoToVeloxConnector::toVeloxTableHandle(
       "Unexpected layout type {}",
       tableHandle.connectorTableLayout->_type);
   return std::make_unique<connector::clp::ClpTableHandle>(
-      tableHandle.connectorId,
-      clpLayout->table.schemaTableName.table,
-      clpLayout->kqlQuery);
+      tableHandle.connectorId, clpLayout->table.schemaTableName.table);
 }
 
 std::unique_ptr<protocol::ConnectorProtocol>

presto-native-execution/presto_cpp/main/connectors/Registration.cpp

@@ -81,8 +81,9 @@ void registerConnectors() {
       std::make_unique<IcebergPrestoToVeloxConnector>(kIcebergConnectorName));
   registerPrestoToVeloxConnector(std::make_unique<TpchPrestoToVeloxConnector>(
       velox::connector::tpch::TpchConnectorFactory::kTpchConnectorName));
-  registerPrestoToVeloxConnector(std::make_unique<ClpPrestoToVeloxConnector>(
-      velox::connector::clp::ClpConnectorFactory::kClpConnectorName));
+  registerPrestoToVeloxConnector(
+      std::make_unique<ClpPrestoToVeloxConnector>(
+          velox::connector::clp::ClpConnectorFactory::kClpConnectorName));
 
   // Presto server uses system catalog or system schema in other catalogs
   // in different places in the code. All these resolve to the SystemConnector.

presto-native-execution/presto_cpp/presto_protocol/connector/clp/presto_protocol_clp.cpp

@@ -81,11 +81,13 @@ void to_json(json& j, const ClpSplit& p) {
   j = json::object();
   j["@type"] = "clp";
   to_json_key(j, "path", p.path, "ClpSplit", "String", "path");
+  to_json_key(j, "kqlQuery", p.kqlQuery, "ClpSplit", "String", "kqlQuery");
 }
 
 void from_json(const json& j, ClpSplit& p) {
   p._type = j["@type"];
   from_json_key(j, "path", p.path, "ClpSplit", "String", "path");
+  from_json_key(j, "kqlQuery", p.kqlQuery, "ClpSplit", "String", "kqlQuery");
 }
 } // namespace facebook::presto::protocol::clp
 namespace facebook::presto::protocol::clp {

presto-native-execution/presto_cpp/presto_protocol/connector/clp/presto_protocol_clp.h

@@ -52,6 +52,7 @@ void from_json(const json& j, ClpColumnHandle& p);
 namespace facebook::presto::protocol::clp {
 struct ClpSplit : public ConnectorSplit {
   String path = {};
+  std::shared_ptr<String> kqlQuery = {};
 
   ClpSplit() noexcept;
 };

presto-native-execution/presto_cpp/presto_protocol/connector/clp/presto_protocol_clp.json

@@ -50,6 +50,14 @@
         "field_text": "String",
         "_N": 1,
         "field_local": true
+      },
+      {
+        "field_type": "Optional<String>",
+        "field_name": "kqlQuery",
+        "field_text": "String",
+        "optional": true,
+        "_N": 2,
+        "field_local": true
       }
     ],
     "subclass": true,