This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 5c3d4bcf6c [spark] prepare MAP selected-key pushdown read type (#8478)
5c3d4bcf6c is described below
commit 5c3d4bcf6c4d0cac6aa26a09c626946f4ad1b8f5
Author: lszskye <[email protected]>
AuthorDate: Wed Jul 8 22:12:22 2026 -0700
[spark] prepare MAP selected-key pushdown read type (#8478)
This PR introduces the preliminary Spark-side and metadata
infrastructure for MAP selected-key pushdown.
The target use case is to allow upper engines, such as Spark, to
describe a read schema like:
`SELECT
id,
attrs['key1'] AS key1_value,
attrs['key2'] AS key2_value
FROM T;`
as a temporary Paimon read type where the MAP field is rewritten to a
ROW field, and the selected MAP keys are recorded in
DataField.description:
`__PAIMON_MAP_SELECTED_KEYS:key1;key2`
This metadata is only intended to exist in the temporary scan/read
schema. It must not be persisted into table schema or catalog schema.
---
.../shredding/MapSelectedKeysMetadataUtils.java | 102 ++++++++
.../MapSelectedKeysMetadataUtilsTest.java | 131 ++++++++++
.../scala/org/apache/paimon/spark/PaimonScan.scala | 1 +
.../optimizer/PushDownMapSelectedKeys.scala | 34 +++
.../scala/org/apache/paimon/spark/PaimonScan.scala | 1 +
.../optimizer/PushDownMapSelectedKeys.scala | 34 +++
.../optimizer/PushDownMapSelectedKeys.scala | 39 +++
.../optimizer/PushDownMapSelectedKeys.scala | 39 +++
.../scala/org/apache/paimon/spark/PaimonScan.scala | 1 +
.../optimizer/PushDownMapSelectedKeys.scala | 289 +++++++++++++++++++++
.../extensions/PaimonSparkSessionExtensions.scala | 2 +
.../org/apache/paimon/spark/read/BaseScan.scala | 21 +-
.../spark/read/MapSelectedKeysPushDownUtils.scala | 71 +++++
.../read/MapSelectedKeysPushDownUtilsTest.scala | 80 ++++++
.../spark/sql/PaimonOptimizationTestBase.scala | 48 +++-
.../optimizer/PushDownMapSelectedKeys.scala | 39 +++
16 files changed, 928 insertions(+), 4 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSelectedKeysMetadataUtils.java
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSelectedKeysMetadataUtils.java
new file mode 100644
index 0000000000..6a4163d673
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSelectedKeysMetadataUtils.java
@@ -0,0 +1,102 @@
+/*
+ * 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.data.shredding;
+
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.RowType;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/**
+ * Utils for marking a temporary read {@link RowType} field as selected-key
MAP output. Uses
+ * description field in DataField to encode selected-key metadata.
+ *
+ * <p>Example: __PAIMON_MAP_SELECTED_KEYS:key1;key2
+ */
+public class MapSelectedKeysMetadataUtils {
+
+ public static final String METADATA_KEY = "__PAIMON_MAP_SELECTED_KEYS:";
+ public static final String KEY_DELIMITER = ";";
+
+ private MapSelectedKeysMetadataUtils() {}
+
+ public static String buildMapSelectedKeysMetadata(List<String> keys) {
+ List<String> normalizedKeys = normalizeKeys(keys);
+ return METADATA_KEY + String.join(KEY_DELIMITER, normalizedKeys);
+ }
+
+ public static boolean isMapSelectedKeysMetadata(@Nullable String
description) {
+ return description != null && description.startsWith(METADATA_KEY);
+ }
+
+ public static boolean isMapSelectedKeysField(DataField field) {
+ return isMapSelectedKeysMetadata(field.description()) && field.type()
instanceof RowType;
+ }
+
+ public static DataField withSelectedKeys(DataField field, DataType
rowType, List<String> keys) {
+ checkArgument(rowType instanceof RowType, "Selected-key MAP read type
must be ROW.");
+ return
field.newType(rowType).newDescription(buildMapSelectedKeysMetadata(keys));
+ }
+
+ public static List<String> selectedKeys(String description) {
+ checkArgument(
+ isMapSelectedKeysMetadata(description),
+ "Invalid selected-key MAP metadata: %s",
+ description);
+ String encoded = description.substring(METADATA_KEY.length());
+
+ String[] parts = encoded.split(KEY_DELIMITER, -1);
+ List<String> keys = new ArrayList<>(parts.length);
+ for (String part : parts) {
+ keys.add(part);
+ }
+ return keys;
+ }
+
+ private static List<String> normalizeKeys(List<String> keys) {
+ checkNotNull(keys, "Selected keys must not be null.");
+ checkArgument(!keys.isEmpty(), "Selected keys must not be empty.");
+
+ Set<String> seenKeys = new LinkedHashSet<>();
+ for (String key : keys) {
+ checkNotNull(key, "Selected key must not be null.");
+ checkArgument(
+ !key.contains(KEY_DELIMITER),
+ "Selected key must not contain '%s': %s",
+ KEY_DELIMITER,
+ key);
+ checkArgument(
+ !key.startsWith(METADATA_KEY),
+ "Selected key must not start with metadata prefix: %s",
+ key);
+ checkArgument(!seenKeys.contains(key), "Selected key must not be
duplicated: %s", key);
+ seenKeys.add(key);
+ }
+ return new ArrayList<>(seenKeys);
+ }
+}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSelectedKeysMetadataUtilsTest.java
b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSelectedKeysMetadataUtilsTest.java
new file mode 100644
index 0000000000..c921ee7528
--- /dev/null
+++
b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSelectedKeysMetadataUtilsTest.java
@@ -0,0 +1,131 @@
+/*
+ * 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.data.shredding;
+
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link MapSelectedKeysMetadataUtils}. */
+class MapSelectedKeysMetadataUtilsTest {
+
+ @Test
+ void testBuildAndParseMetadata() {
+ String metadata =
+ MapSelectedKeysMetadataUtils.buildMapSelectedKeysMetadata(
+ Arrays.asList("key1", "key2"));
+
+ assertThat(metadata).isEqualTo("__PAIMON_MAP_SELECTED_KEYS:key1;key2");
+
assertThat(MapSelectedKeysMetadataUtils.isMapSelectedKeysMetadata(metadata)).isTrue();
+ assertThat(MapSelectedKeysMetadataUtils.selectedKeys(metadata))
+ .containsExactly("key1", "key2");
+ }
+
+ @Test
+ void testBuildAndParseMetadataWithEmptyKey() {
+ String metadata =
+ MapSelectedKeysMetadataUtils.buildMapSelectedKeysMetadata(
+ Arrays.asList("key1", "", "key2"));
+
+ String expectedMetadata = String.join(";",
"__PAIMON_MAP_SELECTED_KEYS:key1", "", "key2");
+ assertThat(metadata).isEqualTo(expectedMetadata);
+ assertThat(MapSelectedKeysMetadataUtils.selectedKeys(metadata))
+ .containsExactly("key1", "", "key2");
+
assertThat(MapSelectedKeysMetadataUtils.selectedKeys("__PAIMON_MAP_SELECTED_KEYS:"))
+ .containsExactly("");
+ }
+
+ @Test
+ void testBuildMetadataRejectsDuplicateKeys() {
+ assertThatThrownBy(
+ () ->
+
MapSelectedKeysMetadataUtils.buildMapSelectedKeysMetadata(
+ Arrays.asList("key1", "key2", "key1")))
+ .hasMessageContaining("Selected key must not be duplicated:
key1");
+ assertThatThrownBy(
+ () ->
+
MapSelectedKeysMetadataUtils.buildMapSelectedKeysMetadata(
+ Arrays.asList("key1", "", "")))
+ .hasMessageContaining("Selected key must not be duplicated");
+ }
+
+ @Test
+ void testWithSelectedKeys() {
+ DataField field =
+ DataTypes.FIELD(
+ 1,
+ "attrs",
+ DataTypes.MAP(DataTypes.STRING(), DataTypes.BIGINT()),
+ "user comment");
+ RowType selectedType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "0", DataTypes.BIGINT()),
+ DataTypes.FIELD(1, "1", DataTypes.BIGINT()));
+
+ DataField rewritten =
+ MapSelectedKeysMetadataUtils.withSelectedKeys(
+ field, selectedType, Arrays.asList("key1", "key2"));
+
+ assertThat(rewritten.id()).isEqualTo(1);
+ assertThat(rewritten.name()).isEqualTo("attrs");
+ assertThat(rewritten.type()).isEqualTo(selectedType);
+
assertThat(rewritten.description()).isEqualTo("__PAIMON_MAP_SELECTED_KEYS:key1;key2");
+
assertThat(MapSelectedKeysMetadataUtils.isMapSelectedKeysField(rewritten)).isTrue();
+ }
+
+ @Test
+ void testErrors() {
+ assertThatThrownBy(
+ () ->
+
MapSelectedKeysMetadataUtils.buildMapSelectedKeysMetadata(
+ Collections.emptyList()))
+ .hasMessageContaining("Selected keys must not be empty.");
+ assertThatThrownBy(
+ () ->
+
MapSelectedKeysMetadataUtils.buildMapSelectedKeysMetadata(
+ Arrays.asList("key1", null)))
+ .hasMessageContaining("Selected key must not be null.");
+ assertThatThrownBy(
+ () ->
+
MapSelectedKeysMetadataUtils.buildMapSelectedKeysMetadata(
+ Arrays.asList("key;1")))
+ .hasMessageContaining("Selected key must not contain ';'");
+ assertThatThrownBy(() ->
MapSelectedKeysMetadataUtils.selectedKeys("invalid"))
+ .hasMessageContaining("Invalid selected-key MAP metadata");
+ assertThatThrownBy(
+ () ->
+ MapSelectedKeysMetadataUtils.withSelectedKeys(
+ DataTypes.FIELD(
+ 1,
+ "attrs",
+ DataTypes.MAP(
+ DataTypes.STRING(),
DataTypes.BIGINT())),
+ DataTypes.BIGINT(),
+ Arrays.asList("key1")))
+ .hasMessageContaining("Selected-key MAP read type must be
ROW.");
+ }
+}
diff --git
a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
index e1fd258186..fab2cfe861 100644
---
a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
+++
b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
@@ -36,5 +36,6 @@ case class PaimonScan(
override val pushedHybridSearch: Option[HybridSearch] = None,
override val pushedFullTextSearch: Option[FullTextSearch] = None,
override val pushedVariantExtractions: Map[Seq[String],
Seq[VariantExtractionInfo]] = Map.empty,
+ override val pushedMapSelectedKeys: Map[String, Seq[String]] = Map.empty,
bucketedScanDisabled: Boolean = true)
extends PaimonBaseScan(table) {}
diff --git
a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
new file mode 100644
index 0000000000..bc4f413e46
--- /dev/null
+++
b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
@@ -0,0 +1,34 @@
+/*
+ * 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.catalyst.optimizer
+
+import org.apache.paimon.spark.PaimonScan
+
+import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation
+
+object PushDownMapSelectedKeys extends PushDownMapSelectedKeysBase {
+
+ override protected def copyDataSourceV2ScanRelation(
+ relation: DataSourceV2ScanRelation,
+ scan: PaimonScan,
+ output: Seq[AttributeReference]): DataSourceV2ScanRelation = {
+ DataSourceV2ScanRelation(relation.relation, scan, output)
+ }
+}
diff --git
a/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
b/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
index 6ecc3dd847..9cc7e409c7 100644
---
a/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
+++
b/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
@@ -40,6 +40,7 @@ case class PaimonScan(
override val pushedHybridSearch: Option[HybridSearch] = None,
override val pushedFullTextSearch: Option[FullTextSearch] = None,
override val pushedVariantExtractions: Map[Seq[String],
Seq[VariantExtractionInfo]] = Map.empty,
+ override val pushedMapSelectedKeys: Map[String, Seq[String]] = Map.empty,
bucketedScanDisabled: Boolean = false)
extends PaimonBaseScan(table)
with SupportsReportPartitioning {
diff --git
a/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
b/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
new file mode 100644
index 0000000000..823c32a302
--- /dev/null
+++
b/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
@@ -0,0 +1,34 @@
+/*
+ * 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.catalyst.optimizer
+
+import org.apache.paimon.spark.PaimonScan
+
+import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation
+
+object PushDownMapSelectedKeys extends PushDownMapSelectedKeysBase {
+
+ override protected def copyDataSourceV2ScanRelation(
+ relation: DataSourceV2ScanRelation,
+ scan: PaimonScan,
+ output: Seq[AttributeReference]): DataSourceV2ScanRelation = {
+ DataSourceV2ScanRelation(relation.relation, scan, output,
relation.keyGroupedPartitioning)
+ }
+}
diff --git
a/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
b/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
new file mode 100644
index 0000000000..88e6347cee
--- /dev/null
+++
b/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
@@ -0,0 +1,39 @@
+/*
+ * 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.catalyst.optimizer
+
+import org.apache.paimon.spark.PaimonScan
+
+import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation
+
+object PushDownMapSelectedKeys extends PushDownMapSelectedKeysBase {
+
+ override protected def copyDataSourceV2ScanRelation(
+ relation: DataSourceV2ScanRelation,
+ scan: PaimonScan,
+ output: Seq[AttributeReference]): DataSourceV2ScanRelation = {
+ DataSourceV2ScanRelation(
+ relation.relation,
+ scan,
+ output,
+ relation.keyGroupedPartitioning,
+ relation.ordering)
+ }
+}
diff --git
a/paimon-spark/paimon-spark-3.5/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
b/paimon-spark/paimon-spark-3.5/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
new file mode 100644
index 0000000000..88e6347cee
--- /dev/null
+++
b/paimon-spark/paimon-spark-3.5/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
@@ -0,0 +1,39 @@
+/*
+ * 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.catalyst.optimizer
+
+import org.apache.paimon.spark.PaimonScan
+
+import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation
+
+object PushDownMapSelectedKeys extends PushDownMapSelectedKeysBase {
+
+ override protected def copyDataSourceV2ScanRelation(
+ relation: DataSourceV2ScanRelation,
+ scan: PaimonScan,
+ output: Seq[AttributeReference]): DataSourceV2ScanRelation = {
+ DataSourceV2ScanRelation(
+ relation.relation,
+ scan,
+ output,
+ relation.keyGroupedPartitioning,
+ relation.ordering)
+ }
+}
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
index 412d92b7f4..1f0b1b75b0 100644
---
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
@@ -46,6 +46,7 @@ case class PaimonScan(
override val pushedHybridSearch: Option[HybridSearch] = None,
override val pushedFullTextSearch: Option[FullTextSearch] = None,
override val pushedVariantExtractions: Map[Seq[String],
Seq[VariantExtractionInfo]] = Map.empty,
+ override val pushedMapSelectedKeys: Map[String, Seq[String]] = Map.empty,
bucketedScanDisabled: Boolean = false)
extends PaimonBaseScan(table)
with SupportsReportPartitioning
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
new file mode 100644
index 0000000000..16db25531f
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
@@ -0,0 +1,289 @@
+/*
+ * 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.catalyst.optimizer
+
+import org.apache.paimon.CoreOptions
+import org.apache.paimon.CoreOptions.MapStorageLayout
+import org.apache.paimon.data.shredding.MapSelectedKeysMetadataUtils
+import org.apache.paimon.data.shredding.MapSharedShreddingUtils
+import org.apache.paimon.spark.PaimonScan
+
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute,
AttributeReference, Expression, ExprId, GetMapValue, GetStructField, Literal,
NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation
+import org.apache.spark.sql.types.{MapType, StringType, StructField,
StructType}
+
+import scala.collection.mutable
+
+/**
+ * Pushes literal-key MAP access on shared-shredding MAP columns into Paimon
scan read type.
+ *
+ * <p>Example: {@code SELECT id, attrs['key1'], attrs['key2'] FROM T} becomes
a scan returning
+ * {@code attrs} as {@code struct<0:valueType,1:valueType>} and a project
reading struct fields.
+ */
+object PushDownMapSelectedKeys extends Rule[LogicalPlan] {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+ plan
+ }
+}
+
+abstract class PushDownMapSelectedKeysBase extends Rule[LogicalPlan] {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case project @ Project(projectList, relation: DataSourceV2ScanRelation) =>
+ relation.scan match {
+ case scan: PaimonScan =>
+ rewriteProject(project, projectList, relation,
scan).getOrElse(project)
+ case _ => project
+ }
+ }
+
+ private def rewriteProject(
+ project: Project,
+ projectList: Seq[NamedExpression],
+ relation: DataSourceV2ScanRelation,
+ scan: PaimonScan): Option[LogicalPlan] = {
+ val candidates = collectCandidates(projectList, scan)
+ if (candidates.isEmpty) {
+ return None
+ }
+
+ val outputByExprId = relation.output.map(a => a.exprId -> a).toMap
+ val selectedMaps = mutable.LinkedHashMap.empty[ExprId, SelectedMap]
+ candidates.foreach {
+ candidate =>
+ outputByExprId.get(candidate.rootExprId).foreach {
+ root =>
+ val selectedMap =
+ selectedMaps.getOrElseUpdate(
+ root.exprId,
+ SelectedMap(
+ root,
+ candidate.fieldName,
+ candidate.mapType,
+ mutable.ArrayBuffer.empty))
+ if (!selectedMap.keys.contains(candidate.key)) {
+ selectedMap.keys.append(candidate.key)
+ }
+ }
+ }
+ if (selectedMaps.isEmpty) {
+ return None
+ }
+
+ val selectedByExprId = selectedMaps.toMap
+ val attrToRewritten = selectedByExprId.map {
+ case (exprId, selected) =>
+ exprId ->
selected.root.withDataType(selected.structType).asInstanceOf[AttributeReference]
+ }
+
+ val rewriteState = RewriteState(selectedByExprId, attrToRewritten)
+ val rewrittenProjectList = projectList.map(rewriteExpression(_,
rewriteState))
+ if (rewriteState.failed) {
+ return None
+ }
+
+ val pushedMapSelectedKeys = scan.pushedMapSelectedKeys ++
selectedMaps.values.map {
+ selected => selected.fieldName -> selected.keys.toSeq
+ }.toMap
+ val rewrittenScan = scan.copy(pushedMapSelectedKeys =
pushedMapSelectedKeys)
+ val rewrittenOutput =
+ relation.output.map(attr => attrToRewritten.getOrElse(attr.exprId, attr))
+ val rewrittenRelation = copyDataSourceV2ScanRelation(relation,
rewrittenScan, rewrittenOutput)
+ Some(project.copy(projectList = rewrittenProjectList, child =
rewrittenRelation))
+ }
+
+ protected def copyDataSourceV2ScanRelation(
+ relation: DataSourceV2ScanRelation,
+ scan: PaimonScan,
+ output: Seq[AttributeReference]): DataSourceV2ScanRelation
+
+ private def collectCandidates(
+ projectList: Seq[NamedExpression],
+ scan: PaimonScan): Seq[MapKeyCandidate] = {
+ val candidates = mutable.ArrayBuffer.empty[MapKeyCandidate]
+ projectList.foreach {
+ expr =>
+ expr.foreach {
+ case mapKeyValueAccess(access) if canPushDown(scan, access) =>
+ candidates.append(
+ MapKeyCandidate(access.root.exprId, access.fieldName,
access.mapType, access.key))
+ case _ =>
+ }
+ }
+ candidates.toSeq
+ }
+
+ private def rewriteExpression(
+ expression: NamedExpression,
+ state: RewriteState): NamedExpression = {
+ expression match {
+ case alias: Alias =>
+ alias.child match {
+ case mapKeyValueAccess(access) if
state.selectedByExprId.contains(access.root.exprId) =>
+ rewriteMapKeyAccess(access, state) match {
+ case Some(rewritten) =>
+
alias.withNewChildren(Seq(rewritten)).asInstanceOf[NamedExpression]
+ case None => expression
+ }
+ case _ =>
+ if (hasSelectedMapReference(alias, state)) {
+ state.failed = true
+ }
+ expression
+ }
+ case attr: Attribute if state.hasRootSelection(attr.exprId) =>
+ state.failed = true
+ expression
+ case _ =>
+ if (hasSelectedMapReference(expression, state)) {
+ state.failed = true
+ }
+ expression
+ }
+ }
+
+ private def rewriteMapKeyAccess(access: MapKeyAccess, state: RewriteState):
Option[Expression] = {
+ val selected = state.selectedByExprId(access.root.exprId)
+ val ordinal = selected.keys.indexOf(access.key)
+ if (ordinal < 0) {
+ state.failed = true
+ None
+ } else {
+ Some(GetStructField(state.attrToRewritten(access.root.exprId), ordinal,
Some(access.key)))
+ }
+ }
+
+ private def hasSelectedMapReference(expression: Expression, state:
RewriteState): Boolean = {
+ var found = false
+ expression.foreach {
+ case mapFieldReference(access) if
state.selectedByExprId.contains(access.root.exprId) =>
+ found = true
+ case _ =>
+ }
+ found
+ }
+
+ private def canPushDown(scan: PaimonScan, access: MapKeyAccess): Boolean = {
+ if (!canEncodeKey(access.key)) {
+ return false
+ }
+
+ access.mapType match {
+ case MapType(StringType, _, _) =>
+ fieldType(scan.table.rowType(), access.fieldName) match {
+ case Some(mapType: org.apache.paimon.types.MapType) if
isStringKeyMap(mapType) =>
+
CoreOptions.fromMap(scan.table.options()).mapStorageLayout(access.fieldName) ==
+ MapStorageLayout.SHARED_SHREDDING
+ case _ => false
+ }
+ case _ => false
+ }
+ }
+
+ private def canEncodeKey(key: String): Boolean = {
+ !key.contains(MapSelectedKeysMetadataUtils.KEY_DELIMITER) &&
+ !key.startsWith(MapSelectedKeysMetadataUtils.METADATA_KEY)
+ }
+
+ private def isStringKeyMap(mapType: org.apache.paimon.types.MapType):
Boolean = {
+ MapSharedShreddingUtils.isShreddingKeyMap(mapType)
+ }
+
+ private object mapKeyValueAccess {
+ def unapply(expression: Expression): Option[MapKeyAccess] = {
+ expression match {
+ case GetMapValue(mapFieldReference(access), StringLiteral(key)) =>
+ Some(access.copy(key = key))
+ case e if e.prettyName == "element_at" && e.children.length == 2 =>
+ (e.children.head, e.children(1)) match {
+ case (mapFieldReference(access), StringLiteral(key)) =>
Some(access.copy(key = key))
+ case _ => None
+ }
+ case _ => None
+ }
+ }
+ }
+
+ private object mapFieldReference {
+ def unapply(expression: Expression): Option[MapKeyAccess] = {
+ expression match {
+ case attr: Attribute =>
+ attr.dataType match {
+ case mapType: MapType => Some(MapKeyAccess(attr, attr.name,
mapType, ""))
+ case _ => None
+ }
+ case _ => None
+ }
+ }
+ }
+
+ private object StringLiteral {
+ def unapply(expression: Expression): Option[String] = {
+ expression match {
+ case Literal(value, StringType) if value != null =>
Some(value.toString)
+ case _ => None
+ }
+ }
+ }
+
+ private def fieldType(
+ rowType: org.apache.paimon.types.RowType,
+ fieldName: String): Option[org.apache.paimon.types.DataType] = {
+ if (!rowType.containsField(fieldName)) {
+ None
+ } else {
+ Some(rowType.getField(fieldName).`type`())
+ }
+ }
+
+ private case class MapKeyCandidate(
+ rootExprId: ExprId,
+ fieldName: String,
+ mapType: MapType,
+ key: String)
+
+ private case class MapKeyAccess(root: Attribute, fieldName: String, mapType:
MapType, key: String)
+
+ private case class SelectedMap(
+ root: Attribute,
+ fieldName: String,
+ mapType: MapType,
+ keys: mutable.ArrayBuffer[String]) {
+ def structType: StructType = {
+ val MapType(_, valueType, _) = mapType
+ StructType(keys.zipWithIndex.map {
+ case (_, ordinal) =>
+ StructField(ordinal.toString, valueType, nullable = true)
+ }.toSeq)
+ }
+ }
+
+ private case class RewriteState(
+ selectedByExprId: Map[ExprId, SelectedMap],
+ attrToRewritten: Map[ExprId, AttributeReference]) {
+ var failed: Boolean = false
+
+ def hasRootSelection(exprId: ExprId): Boolean = {
+ selectedByExprId.contains(exprId)
+ }
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
index 61481e201c..388889bbea 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
@@ -101,6 +101,8 @@ class PaimonSparkSessionExtensions extends
(SparkSessionExtensions => Unit) {
// optimization rules
extensions.injectOptimizerRule(spark => ReplacePaimonFunctions(spark))
extensions.injectOptimizerRule(_ =>
OptimizeMetadataOnlyDeleteFromPaimonTable)
+ // TODO: Enable MAP selected-key pushdown after core reader supports
+ // __PAIMON_MAP_SELECTED_KEYS read type.
extensions.injectOptimizerRule(_ => MergePaimonScalarSubqueries)
extensions.injectOptimizerRule(_ => PushDownLateralVectorSearchFilter)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala
index b36a0f1f02..088b785c7b 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala
@@ -54,6 +54,7 @@ trait BaseScan extends Scan with SupportsReportStatistics
with Logging {
def pushedHybridSearch: Option[HybridSearch] = None
def pushedFullTextSearch: Option[FullTextSearch] = None
def pushedVariantExtractions: Map[Seq[String], Seq[VariantExtractionInfo]] =
Map.empty
+ def pushedMapSelectedKeys: Map[String, Seq[String]] = Map.empty
// Runtime push down
val pushedRuntimePartitionFilters: ListBuffer[PartitionPredicate] =
ListBuffer.empty
@@ -80,7 +81,10 @@ trait BaseScan extends Scan with SupportsReportStatistics
with Logging {
}
}
- /** Pruned read RowType, with variant fields rewritten if variant pushdown
was accepted. */
+ /**
+ * Pruned read RowType, with variant fields rewritten if variant pushdown
was accepted, with
+ * accepted nested field pushdowns rewritten.
+ */
private[paimon] val (readTableRowType, metadataFields) = {
requiredSchema.fields.foreach(f => checkMetadataColumn(f.name))
val (_requiredTableFields, _metadataFields) =
@@ -90,7 +94,10 @@ trait BaseScan extends Scan with SupportsReportStatistics
with Logging {
val withVariants =
if (pushedVariantExtractions.isEmpty) pruned
else VariantPushDownUtils.rewriteRowType(pruned,
pushedVariantExtractions)
- (withVariants, _metadataFields)
+ val withMapSelectedKeys =
+ if (pushedMapSelectedKeys.isEmpty) withVariants
+ else MapSelectedKeysPushDownUtils.rewriteRowType(withVariants,
pushedMapSelectedKeys)
+ (withMapSelectedKeys, _metadataFields)
}
private def checkMetadataColumn(fieldName: String): Unit = {
@@ -198,6 +205,13 @@ trait BaseScan extends Scan with SupportsReportStatistics
with Logging {
.describeRewrittenRowType(readTableRowType)
.map(s => s", PushedVariants: [$s]")
.getOrElse("")
+ val pushedMapSelectedKeysStr =
+ if (pushedMapSelectedKeys.isEmpty) ""
+ else
+ MapSelectedKeysPushDownUtils
+ .describeRewrittenRowType(readTableRowType)
+ .map(s => s", PushedMapSelectedKeys: [$s]")
+ .getOrElse("")
s"${getClass.getSimpleName}: [${table.name}]" +
pushedPartitionFiltersStr +
pushedRuntimePartitionFiltersStr +
@@ -207,6 +221,7 @@ trait BaseScan extends Scan with SupportsReportStatistics
with Logging {
pushedVectorSearch.map(vs => s", VectorSearch: [$vs]").getOrElse("") +
pushedHybridSearch.map(hs => s", HybridSearch: [$hs]").getOrElse("") +
pushedFullTextSearch.map(fts => s", FullTextSearch:
[$fts]").getOrElse("") +
- pushedVariantsStr
+ pushedVariantsStr +
+ pushedMapSelectedKeysStr
}
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/MapSelectedKeysPushDownUtils.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/MapSelectedKeysPushDownUtils.scala
new file mode 100644
index 0000000000..bb131d2e78
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/MapSelectedKeysPushDownUtils.scala
@@ -0,0 +1,71 @@
+/*
+ * 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.read
+
+import org.apache.paimon.data.shredding.MapSelectedKeysMetadataUtils
+import org.apache.paimon.types.{DataField, MapType, RowType}
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+
+/** Shared-shredding MAP selected-key pushdown business logic. */
+object MapSelectedKeysPushDownUtils {
+
+ /** Replace accepted top-level MAP fields with selected-key ROW fields in
the read type. */
+ def rewriteRowType(rt: RowType, accepted: Map[String, Seq[String]]): RowType
= {
+ val newFields = rt.getFields.asScala.map(f => rewriteField(f,
accepted)).asJava
+ new RowType(rt.isNullable, newFields)
+ }
+
+ private def rewriteField(field: DataField, accepted: Map[String,
Seq[String]]): DataField = {
+ accepted.get(field.name()) match {
+ case Some(keys) =>
+ field.`type`() match {
+ case mapType: MapType =>
+ val selectedFields = keys.zipWithIndex.map {
+ case (_, ordinal) =>
+ new DataField(ordinal, ordinal.toString,
mapType.getValueType.copy(true))
+ }.asJava
+ MapSelectedKeysMetadataUtils.withSelectedKeys(
+ field,
+ new RowType(field.`type`().isNullable, selectedFields),
+ keys.asJava)
+ case _ => field
+ }
+ case None => field
+ }
+ }
+
+ /** "col=[key1,key2]" rendering for `Scan.description()`. */
+ def describeRewrittenRowType(rt: RowType): Option[String] = {
+ val parts = mutable.ArrayBuffer.empty[String]
+ collectSelectedKeys(rt, parts)
+ if (parts.isEmpty) None else Some(parts.mkString(", "))
+ }
+
+ private def collectSelectedKeys(rt: RowType, out:
mutable.ArrayBuffer[String]): Unit = {
+ rt.getFields.asScala.foreach {
+ field =>
+ if (MapSelectedKeysMetadataUtils.isMapSelectedKeysField(field)) {
+ val keys =
MapSelectedKeysMetadataUtils.selectedKeys(field.description()).asScala
+ out.append(s"${field.name()}=[${keys.mkString(",")}]")
+ }
+ }
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/read/MapSelectedKeysPushDownUtilsTest.scala
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/read/MapSelectedKeysPushDownUtilsTest.scala
new file mode 100644
index 0000000000..d3019baa59
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/read/MapSelectedKeysPushDownUtilsTest.scala
@@ -0,0 +1,80 @@
+/*
+ * 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.read
+
+import org.apache.paimon.data.shredding.MapSelectedKeysMetadataUtils
+import org.apache.paimon.types.{DataTypes, MapType, RowType}
+
+import org.scalatest.funsuite.AnyFunSuite
+
+import scala.collection.JavaConverters._
+
+/** Tests for {@link MapSelectedKeysPushDownUtils}. */
+class MapSelectedKeysPushDownUtilsTest extends AnyFunSuite {
+
+ test("rewrite map field with selected keys") {
+ val rowType = DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(1, "attrs", DataTypes.MAP(DataTypes.STRING(),
DataTypes.BIGINT().notNull())),
+ DataTypes.FIELD(2, "tags", DataTypes.MAP(DataTypes.STRING(),
DataTypes.STRING()))
+ )
+
+ val rewritten = MapSelectedKeysPushDownUtils.rewriteRowType(
+ rowType,
+ Map("attrs" -> Seq("key1", "key2"))
+ )
+
+ assert(rewritten.getField("id").`type`() == DataTypes.INT())
+ assert(rewritten.getField("tags").`type`().isInstanceOf[MapType])
+
+ val attrs = rewritten.getField("attrs")
+ assert(attrs.description() == "__PAIMON_MAP_SELECTED_KEYS:key1;key2")
+ assert(MapSelectedKeysMetadataUtils.isMapSelectedKeysField(attrs))
+ assert(
+ MapSelectedKeysMetadataUtils.selectedKeys(attrs.description()).asScala
== Seq("key1", "key2"))
+
+ val attrsRowType = attrs.`type`().asInstanceOf[RowType]
+ assert(attrsRowType.getFieldNames.asScala == Seq("0", "1"))
+ assert(attrsRowType.getField("0").`type`() == DataTypes.BIGINT())
+ assert(attrsRowType.getField("1").`type`() == DataTypes.BIGINT())
+ assert(attrsRowType.getField("0").`type`().isNullable)
+ assert(attrsRowType.getField("1").`type`().isNullable)
+ }
+
+ test("ignore nested map field with selected keys") {
+ val nestedType = DataTypes.ROW(
+ DataTypes.FIELD(2, "name", DataTypes.STRING()),
+ DataTypes.FIELD(3, "attrs", DataTypes.MAP(DataTypes.STRING(),
DataTypes.DOUBLE()))
+ )
+ val rowType = DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(1, "profile", nestedType)
+ )
+
+ val rewritten = MapSelectedKeysPushDownUtils.rewriteRowType(
+ rowType,
+ Map("profile.attrs" -> Seq("key1", "key2"))
+ )
+
+ val profile = rewritten.getField("profile").`type`().asInstanceOf[RowType]
+ assert(profile.getField("name").`type`() == DataTypes.STRING())
+ assert(profile.getField("attrs").description() == null)
+ assert(profile.getField("attrs").`type`().isInstanceOf[MapType])
+ }
+}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonOptimizationTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonOptimizationTestBase.scala
index 178006ff1a..0550e1e4e6 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonOptimizationTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonOptimizationTestBase.scala
@@ -19,9 +19,10 @@
package org.apache.paimon.spark.sql
import org.apache.paimon.Snapshot.CommitKind
-import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.spark.{PaimonScan, PaimonSparkTestBase}
import org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionHelper
import org.apache.paimon.spark.catalyst.optimizer.MergePaimonScalarSubqueries
+import org.apache.paimon.spark.catalyst.optimizer.PushDownMapSelectedKeys
import org.apache.paimon.spark.execution.TruncatePaimonTableWithFilterExec
import org.apache.spark.sql.{DataFrame, PaimonUtils, Row}
@@ -29,6 +30,7 @@ import org.apache.spark.sql.catalyst.expressions.{Attribute,
CreateNamedStruct,
import org.apache.spark.sql.catalyst.plans.logical.{CTERelationDef,
LogicalPlan, OneRowRelation, WithCTE}
import org.apache.spark.sql.catalyst.rules.RuleExecutor
import org.apache.spark.sql.execution.CommandResultExec
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation
import org.apache.spark.sql.functions._
import org.junit.jupiter.api.Assertions
@@ -118,6 +120,44 @@ abstract class PaimonOptimizationTestBase extends
PaimonSparkTestBase with Expre
}
}
+ test("Paimon Optimization: map selected-key pushdown only supports
shared-shredding map") {
+ withTable("T") {
+ spark.sql("CREATE TABLE T (id INT, attrs MAP<STRING, BIGINT>)")
+
+ val normalMapScan = mapSelectedKeysPaimonScan("SELECT attrs['key1'] FROM
T")
+ Assertions.assertTrue(normalMapScan.pushedMapSelectedKeys.isEmpty)
+ }
+
+ withTable("T") {
+ spark.sql("CREATE TABLE T (id INT, profile STRUCT<attrs: MAP<STRING,
BIGINT>>)")
+
+ val nestedMapScan = mapSelectedKeysPaimonScan("SELECT
profile.attrs['key1'] FROM T")
+ Assertions.assertTrue(nestedMapScan.pushedMapSelectedKeys.isEmpty)
+ }
+
+ withTable("T") {
+ spark.sql("""
+ |CREATE TABLE T (id INT, attrs MAP<STRING, BIGINT>)
+ |TBLPROPERTIES (
+ | 'fields.attrs.map.storage-layout' = 'shared-shredding'
+ |)
+ |""".stripMargin)
+
+ val sharedShreddingMapScan =
+ mapSelectedKeysPaimonScan("SELECT attrs['key1'], attrs['key2'] FROM T")
+ Assertions.assertEquals(
+ Map("attrs" -> Seq("key1", "key2")),
+ sharedShreddingMapScan.pushedMapSelectedKeys)
+
+ val delimiterKeyScan = mapSelectedKeysPaimonScan("SELECT attrs['a;b']
FROM T")
+ Assertions.assertTrue(delimiterKeyScan.pushedMapSelectedKeys.isEmpty)
+
+ val metadataPrefixKeyScan =
+ mapSelectedKeysPaimonScan("SELECT
attrs['__PAIMON_MAP_SELECTED_KEYS:key1'] FROM T")
+
Assertions.assertTrue(metadataPrefixKeyScan.pushedMapSelectedKeys.isEmpty)
+ }
+ }
+
test(s"Paimon Optimization: optimize metadata only delete") {
for (useV2Write <- Seq("true", "false")) {
withSparkSQLConf("spark.paimon.write.use-v2-write" -> useV2Write) {
@@ -233,6 +273,12 @@ abstract class PaimonOptimizationTestBase extends
PaimonSparkTestBase with Expre
def extractorExpression(cteIndex: Int, output: Seq[Attribute], fieldIndex:
Int): NamedExpression
+ private def mapSelectedKeysPaimonScan(sqlText: String): PaimonScan = {
+
PushDownMapSelectedKeys(sql(sqlText).queryExecution.optimizedPlan).collectFirst
{
+ case relation: DataSourceV2ScanRelation =>
relation.scan.asInstanceOf[PaimonScan]
+ }.get
+ }
+
def checkTruncatePaimonTable(df: DataFrame): Unit = {
val plan =
df.queryExecution.executedPlan.asInstanceOf[CommandResultExec].commandPhysicalPlan
assert(plan.isInstanceOf[TruncatePaimonTableWithFilterExec])
diff --git
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
new file mode 100644
index 0000000000..88e6347cee
--- /dev/null
+++
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownMapSelectedKeys.scala
@@ -0,0 +1,39 @@
+/*
+ * 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.catalyst.optimizer
+
+import org.apache.paimon.spark.PaimonScan
+
+import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation
+
+object PushDownMapSelectedKeys extends PushDownMapSelectedKeysBase {
+
+ override protected def copyDataSourceV2ScanRelation(
+ relation: DataSourceV2ScanRelation,
+ scan: PaimonScan,
+ output: Seq[AttributeReference]): DataSourceV2ScanRelation = {
+ DataSourceV2ScanRelation(
+ relation.relation,
+ scan,
+ output,
+ relation.keyGroupedPartitioning,
+ relation.ordering)
+ }
+}
