You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2023/12/01 09:55:55 UTC
(incubator-paimon) branch release-0.6 updated (29ac8def1 -> d15d7d165)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a change to branch release-0.6
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
from 29ac8def1 [hive] HiveCatalog infer warehouse from hadoop Conf (#2432)
new 555ff8fe9 [core] Fix migration action and procedure name (#2436)
new 63262b63b [spark] Supports runtime filter (#2421)
new d15d7d165 [core] MergeFunctionWrapper should handle return null (#2438)
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/content/migration/migration-from-hive.md | 16 +--
.../compact/FirstRowMergeTreeCompactRewriter.java | 5 +-
.../LookupChangelogMergeFunctionWrapper.java | 5 +-
.../paimon/table/source/ReadBuilderImpl.java | 7 +-
.../LookupChangelogMergeFunctionWrapperTest.java | 27 +++++
.../flink/action/MigrateTableActionFactory.java | 12 +-
.../{PaimonScanBuilder.scala => PaimonScan.scala} | 4 +-
.../java/org/apache/paimon/spark/SparkScan.java | 124 ---------------------
.../spark/procedure/MigrateFileProcedure.java | 4 +-
.../spark/procedure/MigrateTableProcedure.java | 6 +-
.../org/apache/paimon/spark/PaimonBaseScan.scala | 72 ++++++++++++
.../paimon/spark/PaimonBaseScanBuilder.scala | 2 +-
.../scala/org/apache/paimon/spark/PaimonScan.scala | 57 ++++++++++
.../main/scala/org/apache/spark/sql/Utils.scala | 5 +
.../spark/procedure/MigrateFileProcedureTest.scala | 4 +-
.../procedure/MigrateTableProcedureTest.scala | 4 +-
.../paimon/spark/sql/PaimonPushDownTest.scala | 45 +++++++-
17 files changed, 246 insertions(+), 153 deletions(-)
copy paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/{PaimonScanBuilder.scala => PaimonScan.scala} (84%)
delete mode 100644 paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkScan.java
create mode 100644 paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
create mode 100644 paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
(incubator-paimon) 01/03: [core] Fix migration action and procedure name (#2436)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.6
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit 555ff8fe9957b19e47a2c09b07c24f3a285d77a3
Author: YeJunHao <41...@users.noreply.github.com>
AuthorDate: Fri Dec 1 15:17:54 2023 +0800
[core] Fix migration action and procedure name (#2436)
---
docs/content/migration/migration-from-hive.md | 16 ++++++++--------
.../paimon/flink/action/MigrateTableActionFactory.java | 12 ++++++------
.../paimon/spark/procedure/MigrateFileProcedure.java | 4 ++--
.../paimon/spark/procedure/MigrateTableProcedure.java | 6 +++---
.../spark/procedure/MigrateFileProcedureTest.scala | 4 ++--
.../spark/procedure/MigrateTableProcedureTest.scala | 4 ++--
6 files changed, 23 insertions(+), 23 deletions(-)
diff --git a/docs/content/migration/migration-from-hive.md b/docs/content/migration/migration-from-hive.md
index 9d88d03a7..aa8b79a18 100644
--- a/docs/content/migration/migration-from-hive.md
+++ b/docs/content/migration/migration-from-hive.md
@@ -71,20 +71,20 @@ If your flink version is below 1.17, you can use flink action to achieve this:
/path/to/paimon-flink-action-{{< version >}}.jar \
migrate_table
--warehouse <warehouse-path> \
---source-table-type hive \
---source-table-id <database.table-name> \
-[--catalog-conf <paimon-catalog-conf> [--catalog-conf <paimon-catalog-conf> ...]] \
-[--table-conf <paimon-table-sink-conf> [--table-conf <paimon-table-sink-conf> ...]]
+--source_type hive \
+--table <database.table-name> \
+[--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]] \
+[--options <paimon-table-conf [,paimon-table-conf ...]> ]
```
Example:
```bash
<FLINK_HOME>/flink run ./paimon-flink-action-0.7-SNAPSHOT.jar migrate_table \
--warehouse /path/to/warehouse \
---catalog-conf uri=thrift://localhost:9083 \
---catalog-conf metastore=hive \
---source-table-type hive \
---source-table-id default.hive_or_paimon \
+--catalog_conf uri=thrift://localhost:9083 \
+--catalog_conf metastore=hive \
+--source_type hive \
+--table default.hive_or_paimon \
```
**Migrate Hive File**
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateTableActionFactory.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateTableActionFactory.java
index 9efbe6cf4..a9dd8a512 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateTableActionFactory.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateTableActionFactory.java
@@ -36,10 +36,10 @@ public class MigrateTableActionFactory implements ActionFactory {
@Override
public Optional<Action> create(MultipleParameterTool params) {
String warehouse = params.get("warehouse");
- String connector = params.get("source-table-type");
- String sourceHiveTable = params.get("source-table-id");
+ String connector = params.get("source_type");
+ String sourceHiveTable = params.get("table");
Map<String, String> catalogConfig = optionalConfigMap(params, "catalog-conf");
- String tableConf = params.get("table-properties");
+ String tableConf = params.get("options");
MigrateTableAction migrateTableAction =
new MigrateTableAction(
@@ -54,9 +54,9 @@ public class MigrateTableActionFactory implements ActionFactory {
System.out.println("Syntax:");
System.out.println(
- " compact --warehouse <warehouse-path> --source-table-type hive "
- + "--source-table-id <database.table_name> "
+ " migrate_table --warehouse <warehouse-path> --source_type hive "
+ + "--table <database.table_name> "
+ "[--catalog-conf <key>=<value] "
- + "[--table-properties <key>=<value>,<key>=<value>,...]");
+ + "[--options <key>=<value>,<key>=<value>,...]");
}
}
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateFileProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateFileProcedure.java
index 5978767ae..8df10ffd5 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateFileProcedure.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateFileProcedure.java
@@ -39,14 +39,14 @@ import static org.apache.spark.sql.types.DataTypes.StringType;
* Migrate file procedure. Usage:
*
* <pre><code>
- * CALL sys.migrate_file(format => 'hive', source_table => 'db.source_tbl', target_table => 'db.target_tbl')
+ * CALL sys.migrate_file(source_type => 'hive', source_table => 'db.source_tbl', target_table => 'db.target_tbl')
* </code></pre>
*/
public class MigrateFileProcedure extends BaseProcedure {
private static final ProcedureParameter[] PARAMETERS =
new ProcedureParameter[] {
- ProcedureParameter.required("format", StringType),
+ ProcedureParameter.required("source_type", StringType),
ProcedureParameter.required("source_table", StringType),
ProcedureParameter.required("target_table", StringType)
};
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateTableProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateTableProcedure.java
index bc15be752..a057999c1 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateTableProcedure.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateTableProcedure.java
@@ -38,7 +38,7 @@ import static org.apache.spark.sql.types.DataTypes.StringType;
* Migrate table procedure. Usage:
*
* <pre><code>
- * CALL sys.migrate_table(format => 'hive', table => 'db.tbl', tblproperties => 'x1=y1,x2=y2')
+ * CALL sys.migrate_table(source_type => 'hive', table => 'db.tbl', options => 'x1=y1,x2=y2')
* </code></pre>
*/
public class MigrateTableProcedure extends BaseProcedure {
@@ -47,9 +47,9 @@ public class MigrateTableProcedure extends BaseProcedure {
private static final ProcedureParameter[] PARAMETERS =
new ProcedureParameter[] {
- ProcedureParameter.required("format", StringType),
+ ProcedureParameter.required("source_type", StringType),
ProcedureParameter.required("table", StringType),
- ProcedureParameter.optional("tblproperties", StringType)
+ ProcedureParameter.optional("options", StringType)
};
private static final StructType OUTPUT_TYPE =
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateFileProcedureTest.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateFileProcedureTest.scala
index 86d96274c..25a7e34d1 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateFileProcedureTest.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateFileProcedureTest.scala
@@ -44,7 +44,7 @@ class MigrateFileProcedureTest extends PaimonHiveTestBase {
spark.sql(s"INSERT INTO paimon_tbl VALUES ('3', 'c', 'p1'), ('4', 'd', 'p2')")
spark.sql(
- s"CALL sys.migrate_file(format => 'hive', source_table => '$hiveDbName.hive_tbl', target_table => '$hiveDbName.paimon_tbl')")
+ s"CALL sys.migrate_file(source_type => 'hive', source_table => '$hiveDbName.hive_tbl', target_table => '$hiveDbName.paimon_tbl')")
checkAnswer(
spark.sql("SELECT * FROM paimon_tbl ORDER BY id"),
@@ -80,7 +80,7 @@ class MigrateFileProcedureTest extends PaimonHiveTestBase {
spark.sql(s"INSERT INTO paimon_tbl VALUES ('3', 'c', 'p1'), ('4', 'd', 'p2')")
spark.sql(
- s"CALL sys.migrate_file(format => 'hive', source_table => '$hiveDbName.hive_tbl', target_table => '$hiveDbName.paimon_tbl')")
+ s"CALL sys.migrate_file(source_type => 'hive', source_table => '$hiveDbName.hive_tbl', target_table => '$hiveDbName.paimon_tbl')")
checkAnswer(
spark.sql("SELECT * FROM paimon_tbl ORDER BY id"),
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateTableProcedureTest.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateTableProcedureTest.scala
index 179543a48..cf1c8a6f7 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateTableProcedureTest.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateTableProcedureTest.scala
@@ -35,7 +35,7 @@ class MigrateTableProcedureTest extends PaimonHiveTestBase {
spark.sql(s"INSERT INTO hive_tbl VALUES ('1', 'a', 'p1'), ('2', 'b', 'p2')")
spark.sql(
- s"CALL sys.migrate_table(format => 'hive', table => '$hiveDbName.hive_tbl', tblproperties => 'file.format=$format')")
+ s"CALL sys.migrate_table(source_type => 'hive', table => '$hiveDbName.hive_tbl', options => 'file.format=$format')")
checkAnswer(
spark.sql(s"SELECT * FROM hive_tbl ORDER BY id"),
@@ -58,7 +58,7 @@ class MigrateTableProcedureTest extends PaimonHiveTestBase {
spark.sql(s"INSERT INTO hive_tbl VALUES ('1', 'a', 'p1'), ('2', 'b', 'p2')")
spark.sql(
- s"CALL sys.migrate_table(format => 'hive', table => '$hiveDbName.hive_tbl', tblproperties => 'file.format=$format')")
+ s"CALL sys.migrate_table(source_type => 'hive', table => '$hiveDbName.hive_tbl', options => 'file.format=$format')")
checkAnswer(
spark.sql(s"SELECT * FROM hive_tbl ORDER BY id"),
(incubator-paimon) 02/03: [spark] Supports runtime filter (#2421)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.6
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit 63262b63bcb2eb615bb3b7fb0aee96b74f494f41
Author: Yann Byron <bi...@gmail.com>
AuthorDate: Fri Dec 1 17:48:07 2023 +0800
[spark] Supports runtime filter (#2421)
---
.../paimon/table/source/ReadBuilderImpl.java | 7 +-
.../scala/org/apache/paimon/spark/PaimonScan.scala | 24 ++++
.../java/org/apache/paimon/spark/SparkScan.java | 124 ---------------------
.../org/apache/paimon/spark/PaimonBaseScan.scala | 72 ++++++++++++
.../paimon/spark/PaimonBaseScanBuilder.scala | 2 +-
.../scala/org/apache/paimon/spark/PaimonScan.scala | 57 ++++++++++
.../main/scala/org/apache/spark/sql/Utils.scala | 5 +
.../paimon/spark/sql/PaimonPushDownTest.scala | 45 +++++++-
8 files changed, 209 insertions(+), 127 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java
index 6805f30e4..1ceb614ec 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java
@@ -19,6 +19,7 @@
package org.apache.paimon.table.source;
import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.table.InnerTable;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Projection;
@@ -61,7 +62,11 @@ public class ReadBuilderImpl implements ReadBuilder {
@Override
public ReadBuilder withFilter(Predicate filter) {
- this.filter = filter;
+ if (this.filter == null) {
+ this.filter = filter;
+ } else {
+ this.filter = PredicateBuilder.and(this.filter, filter);
+ }
return this;
}
diff --git a/paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/PaimonScan.scala b/paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
new file mode 100644
index 000000000..f4cc95b63
--- /dev/null
+++ b/paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.paimon.spark
+
+import org.apache.paimon.table.Table
+import org.apache.paimon.table.source.ReadBuilder
+
+case class PaimonScan(table: Table, readBuilder: ReadBuilder)
+ extends PaimonBaseScan(table, readBuilder)
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkScan.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkScan.java
deleted file mode 100644
index 891420beb..000000000
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkScan.java
+++ /dev/null
@@ -1,124 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.spark;
-
-import org.apache.paimon.spark.sources.PaimonMicroBatchStream;
-import org.apache.paimon.table.DataTable;
-import org.apache.paimon.table.Table;
-import org.apache.paimon.table.source.ReadBuilder;
-import org.apache.paimon.table.source.Split;
-
-import org.apache.spark.sql.connector.read.Batch;
-import org.apache.spark.sql.connector.read.Scan;
-import org.apache.spark.sql.connector.read.Statistics;
-import org.apache.spark.sql.connector.read.SupportsReportStatistics;
-import org.apache.spark.sql.connector.read.streaming.MicroBatchStream;
-import org.apache.spark.sql.types.StructType;
-
-import java.util.List;
-import java.util.OptionalLong;
-
-/**
- * A Spark {@link Scan} for paimon.
- *
- * <p>TODO Introduce a SparkRFScan to implement SupportsRuntimeFiltering.
- */
-public class SparkScan implements Scan, SupportsReportStatistics {
-
- private final Table table;
- private final ReadBuilder readBuilder;
-
- private List<Split> splits;
-
- public SparkScan(Table table, ReadBuilder readBuilder) {
- this.table = table;
- this.readBuilder = readBuilder;
- }
-
- @Override
- public String description() {
- // TODO add filters
- return String.format("paimon(%s)", readBuilder.tableName());
- }
-
- @Override
- public StructType readSchema() {
- return SparkTypeUtils.fromPaimonRowType(readBuilder.readType());
- }
-
- @Override
- public Batch toBatch() {
- return new PaimonBatch(splits().toArray(new Split[0]), readBuilder);
- }
-
- @Override
- public MicroBatchStream toMicroBatchStream(String checkpointLocation) {
- return new PaimonMicroBatchStream((DataTable) table, readBuilder, checkpointLocation);
- }
-
- protected List<Split> splits() {
- if (splits == null) {
- splits = readBuilder.newScan().plan().splits();
- }
- return splits;
- }
-
- @Override
- public Statistics estimateStatistics() {
- long rowCount = 0L;
-
- for (Split split : splits()) {
- rowCount += split.rowCount();
- }
-
- final long numRows = rowCount;
- final long sizeInBytes = readSchema().defaultSize() * numRows;
-
- return new Statistics() {
- @Override
- public OptionalLong sizeInBytes() {
- return OptionalLong.of(sizeInBytes);
- }
-
- @Override
- public OptionalLong numRows() {
- return OptionalLong.of(numRows);
- }
- };
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
-
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
-
- SparkScan that = (SparkScan) o;
- return readBuilder.equals(that.readBuilder);
- }
-
- @Override
- public int hashCode() {
- return readBuilder.hashCode();
- }
-}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
new file mode 100644
index 000000000..556d1bd8b
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.paimon.spark
+
+import org.apache.paimon.spark.sources.PaimonMicroBatchStream
+import org.apache.paimon.table.{DataTable, Table}
+import org.apache.paimon.table.source.{ReadBuilder, Split}
+
+import org.apache.spark.sql.connector.read.{Batch, Scan, Statistics, SupportsReportStatistics}
+import org.apache.spark.sql.connector.read.streaming.MicroBatchStream
+import org.apache.spark.sql.types.StructType
+
+import java.util.OptionalLong
+
+import scala.collection.JavaConverters._
+
+abstract class PaimonBaseScan(table: Table, readBuilder: ReadBuilder)
+ extends Scan
+ with SupportsReportStatistics {
+
+ protected var splits: Array[Split] = _
+
+ override def description(): String = {
+ s"paimon(${readBuilder.tableName()})"
+ }
+
+ override def readSchema(): StructType = {
+ SparkTypeUtils.fromPaimonRowType(readBuilder.readType())
+ }
+
+ override def toBatch: Batch = {
+ PaimonBatch(getSplits, readBuilder)
+ }
+
+ override def toMicroBatchStream(checkpointLocation: String): MicroBatchStream = {
+ new PaimonMicroBatchStream(table.asInstanceOf[DataTable], readBuilder, checkpointLocation)
+ }
+
+ override def estimateStatistics(): Statistics = {
+ val rowCount = getSplits.map(_.rowCount).sum
+ val scannedTotalSize = rowCount * readSchema().defaultSize
+
+ new Statistics {
+ override def sizeInBytes(): OptionalLong = OptionalLong.of(scannedTotalSize)
+
+ override def numRows(): OptionalLong = OptionalLong.of(rowCount)
+ }
+ }
+
+ private def getSplits: Array[Split] = {
+ if (splits == null) {
+ splits = readBuilder.newScan().plan().splits().asScala.toArray
+ }
+ splits
+ }
+
+}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScanBuilder.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScanBuilder.scala
index d7d4535ea..831b848bc 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScanBuilder.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScanBuilder.scala
@@ -49,7 +49,7 @@ abstract class PaimonBaseScanBuilder(table: Table)
}
override def build(): Scan = {
- new SparkScan(table, getReadBuilder());
+ new PaimonScan(table, getReadBuilder());
}
/**
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScan.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
new file mode 100644
index 000000000..9e76aecbc
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.paimon.spark
+
+import org.apache.paimon.table.Table
+import org.apache.paimon.table.source.ReadBuilder
+
+import org.apache.spark.sql.Utils.fieldReference
+import org.apache.spark.sql.connector.expressions.NamedReference
+import org.apache.spark.sql.connector.read.SupportsRuntimeFiltering
+import org.apache.spark.sql.sources.{Filter, In}
+
+import scala.collection.JavaConverters._
+
+case class PaimonScan(table: Table, readBuilder: ReadBuilder)
+ extends PaimonBaseScan(table, readBuilder)
+ with SupportsRuntimeFiltering {
+
+ override def filterAttributes(): Array[NamedReference] = {
+ val requiredFields = readBuilder.readType().getFieldNames.asScala
+ table
+ .partitionKeys()
+ .asScala
+ .toArray
+ .filter(requiredFields.contains)
+ .map(fieldReference)
+ }
+
+ override def filter(filters: Array[Filter]): Unit = {
+ val converter = new SparkFilterConverter(table.rowType())
+ val partitionFilter = filters.flatMap {
+ case in @ In(attr, _) if table.partitionKeys().contains(attr) =>
+ Some(converter.convert(in))
+ case _ => None
+ }
+ if (partitionFilter.nonEmpty) {
+ readBuilder.withFilter(partitionFilter.head)
+ splits = readBuilder.newScan().plan().splits().asScala.toArray
+ }
+ }
+
+}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/Utils.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/Utils.scala
index feb8815f9..058664270 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/Utils.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/Utils.scala
@@ -19,6 +19,7 @@ package org.apache.spark.sql
import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.connector.expressions.{FieldReference, NamedReference}
import org.apache.spark.sql.execution.datasources.DataSourceStrategy
import org.apache.spark.sql.sources.Filter
@@ -58,4 +59,8 @@ object Utils {
supportNestedPredicatePushdown: Boolean): Option[Filter] = {
DataSourceStrategy.translateFilter(predicate, supportNestedPredicatePushdown)
}
+
+ def fieldReference(name: String): NamedReference = {
+ FieldReference.column(name)
+ }
}
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/PaimonPushDownTest.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/PaimonPushDownTest.scala
index 9d3532adf..472f41118 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/PaimonPushDownTest.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/PaimonPushDownTest.scala
@@ -22,13 +22,16 @@ import org.apache.paimon.table.source.DataSplit
import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.expressions.{AttributeReference, EqualTo, Expression, Literal}
import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.trees.TreePattern.DYNAMIC_PRUNING_SUBQUERY
import org.apache.spark.sql.connector.read.{ScanBuilder, SupportsPushDownLimit}
import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.junit.jupiter.api.Assertions
import scala.collection.JavaConverters._
-class SparkPushDownTest extends PaimonSparkTestBase {
+class PaimonPushDownTest extends PaimonSparkTestBase {
+
+ import testImplicits._
test(s"Paimon push down: apply partition filter push down with non-partitioned table") {
spark.sql(s"""
@@ -129,6 +132,46 @@ class SparkPushDownTest extends PaimonSparkTestBase {
Assertions.assertFalse(scanBuilder.asInstanceOf[SupportsPushDownLimit].pushLimit(1))
}
+ test("Paimon pushDown: runtime filter") {
+ withTable("source", "t") {
+ Seq((1L, "x1", "2023"), (2L, "x2", "2023"), (5L, "x5", "2025"), (6L, "x6", "2026"))
+ .toDF("a", "b", "pt")
+ .createOrReplaceTempView("source")
+
+ spark.sql("""
+ |CREATE TABLE t (id INT, name STRING, pt STRING) PARTITIONED BY (pt)
+ |""".stripMargin)
+
+ spark.sql(
+ """
+ |INSERT INTO t VALUES (1, "a", "2023"), (3, "c", "2023"), (5, "e", "2025"), (7, "g", "2027")
+ |""".stripMargin)
+
+ val df1 = spark.sql("""
+ |SELECT t.id, t.name, source.b FROM source join t
+ |ON source.pt = t.pt AND source.pt = '2023'
+ |ORDER BY t.id, source.b
+ |""".stripMargin)
+ val qe1 = df1.queryExecution
+ Assertions.assertFalse(qe1.analyzed.containsPattern(DYNAMIC_PRUNING_SUBQUERY))
+ Assertions.assertTrue(qe1.optimizedPlan.containsPattern(DYNAMIC_PRUNING_SUBQUERY))
+ Assertions.assertTrue(qe1.sparkPlan.containsPattern(DYNAMIC_PRUNING_SUBQUERY))
+ checkAnswer(
+ df1,
+ Row(1, "a", "x1") :: Row(1, "a", "x2") :: Row(3, "c", "x1") :: Row(3, "c", "x2") :: Nil)
+
+ val df2 = spark.sql("""
+ |SELECT t.*, source.b FROM source join t
+ |ON source.a = t.id AND source.pt = t.pt AND source.a > 3
+ |""".stripMargin)
+ val qe2 = df1.queryExecution
+ Assertions.assertFalse(qe2.analyzed.containsPattern(DYNAMIC_PRUNING_SUBQUERY))
+ Assertions.assertTrue(qe2.optimizedPlan.containsPattern(DYNAMIC_PRUNING_SUBQUERY))
+ Assertions.assertTrue(qe2.sparkPlan.containsPattern(DYNAMIC_PRUNING_SUBQUERY))
+ checkAnswer(df2, Row(5, "e", "2025", "x5") :: Nil)
+ }
+ }
+
private def getScanBuilder(tableName: String = "T"): ScanBuilder = {
new SparkTable(loadTable(tableName))
.newScanBuilder(CaseInsensitiveStringMap.empty())
(incubator-paimon) 03/03: [core] MergeFunctionWrapper should handle return null (#2438)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.6
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit d15d7d165c74f9ccacffe02af993ad0ea479648e
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Fri Dec 1 17:51:04 2023 +0800
[core] MergeFunctionWrapper should handle return null (#2438)
---
.../compact/FirstRowMergeTreeCompactRewriter.java | 5 +++-
.../LookupChangelogMergeFunctionWrapper.java | 5 +++-
.../LookupChangelogMergeFunctionWrapperTest.java | 27 ++++++++++++++++++++++
3 files changed, 35 insertions(+), 2 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FirstRowMergeTreeCompactRewriter.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FirstRowMergeTreeCompactRewriter.java
index d79cef325..9ae4da9b9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FirstRowMergeTreeCompactRewriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FirstRowMergeTreeCompactRewriter.java
@@ -126,7 +126,10 @@ public class FirstRowMergeTreeCompactRewriter extends ChangelogMergeTreeRewriter
public ChangelogResult getResult() {
reusedResult.reset();
KeyValue result = mergeFunction.getResult();
- checkArgument(result != null);
+ if (result == null) {
+ return reusedResult;
+ }
+
if (contains.test(result.key())) {
// empty
return reusedResult;
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapper.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapper.java
index e405d33cb..717a32727 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapper.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapper.java
@@ -86,7 +86,10 @@ public class LookupChangelogMergeFunctionWrapper implements MergeFunctionWrapper
reusedResult.reset();
KeyValue result = mergeFunction.getResult();
- checkArgument(result != null);
+ if (result == null) {
+ return reusedResult;
+ }
+
KeyValue highLevel = mergeFunction.highLevel;
boolean containLevel0 = mergeFunction.containLevel0;
diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapperTest.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapperTest.java
index 07c93e1a7..df51eecff 100644
--- a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapperTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapperTest.java
@@ -18,6 +18,7 @@
package org.apache.paimon.mergetree.compact;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.KeyValue;
import org.apache.paimon.codegen.RecordEqualiser;
import org.apache.paimon.data.InternalRow;
@@ -25,6 +26,7 @@ import org.apache.paimon.data.InternalRow.FieldGetter;
import org.apache.paimon.mergetree.compact.aggregate.AggregateMergeFunction;
import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
import org.apache.paimon.mergetree.compact.aggregate.FieldSumAgg;
+import org.apache.paimon.options.Options;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.IntType;
@@ -35,6 +37,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -365,4 +368,28 @@ public class LookupChangelogMergeFunctionWrapperTest {
kv = result.result();
assertThat(kv).isNull();
}
+
+ @Test
+ public void testPartialUpdateIgnoreDelete() {
+ Options options = new Options();
+ options.set(CoreOptions.PARTIAL_UPDATE_IGNORE_DELETE, true);
+ LookupChangelogMergeFunctionWrapper function =
+ new LookupChangelogMergeFunctionWrapper(
+ LookupMergeFunction.wrap(
+ PartialUpdateMergeFunction.factory(
+ options,
+ DataTypes.ROW(DataTypes.INT()),
+ Collections.singletonList("f0")),
+ RowType.of(DataTypes.INT()),
+ RowType.of(DataTypes.INT())),
+ key -> null,
+ EQUALISER,
+ false);
+
+ function.reset();
+ function.add(new KeyValue().replace(row(1), 1, DELETE, row(1)).setLevel(2));
+ ChangelogResult result = function.getResult();
+ assertThat(result).isNotNull();
+ assertThat(result.result()).isNull();
+ }
}