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 bbad1a0f3a [core][spark] Support pluggable fixed-bucket functions
(#10662)
bbad1a0f3a is described below
commit bbad1a0f3a62d07eb02f15fb0e1a576ce1740fc6
Author: Luo Zhengjie <[email protected]>
AuthorDate: Tue Oct 27 13:12:55 2026 +0800
[core][spark] Support pluggable fixed-bucket functions (#10662)
---
docs/generated/core_configuration.html | 6 +
.../main/java/org/apache/paimon/CoreOptions.java | 15 +++
.../org/apache/paimon/AppendOnlyFileStore.java | 6 +-
.../java/org/apache/paimon/KeyValueFileStore.java | 2 +-
.../org/apache/paimon/bucket/BucketFunction.java | 18 +++
.../paimon/bucket/BucketFunctionFactory.java | 45 ++++++++
.../paimon/operation/BucketSelectConverter.java | 31 ++++-
.../apache/paimon/operation/BucketSelector.java | 36 +++++-
.../org/apache/paimon/schema/SchemaValidation.java | 8 ++
.../paimon/bucket/BucketFunctionProviderTest.java | 46 ++++++++
.../bucket/ConcurrentBucketFunctionFactory.java | 67 +++++++++++
.../paimon/bucket/TestBucketFunctionFactory.java | 38 +++++++
.../paimon/operation/BucketSelectorTest.java | 23 ++++
.../apache/paimon/table/BucketFilterScanTest.java | 56 +++++++++
.../table/sink/FixedBucketRowKeyExtractorTest.java | 29 +++++
.../services/org.apache.paimon.factories.Factory | 2 +
.../scala/org/apache/paimon/spark/PaimonScan.scala | 7 +-
.../spark/sql/CustomBucketFunction33Test.scala | 21 ++++
.../scala/org/apache/paimon/spark/PaimonScan.scala | 3 +-
.../apache/paimon/spark/PaimonSparkTableBase.scala | 1 +
.../paimon/spark/commands/PaimonSparkWriter.scala | 5 +-
.../paimon/spark/ShiftedBucketFunctionFactory.java | 43 +++++++
.../services/org.apache.paimon.factories.Factory | 18 +++
.../spark/sql/CustomBucketFunctionTest.scala | 125 +++++++++++++++++++++
24 files changed, 634 insertions(+), 17 deletions(-)
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index 05585ef5a4..ef54534e46 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -140,6 +140,12 @@ under the License.
<td>Boolean</td>
<td>Whether to ignore the order of the buckets when reading data
from an append-only table.</td>
</tr>
+ <tr>
+ <td><h5>bucket-function.provider</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Optional identifier of a BucketFunctionFactory discovered from
the classpath. When configured, the provider is used by both writers and bucket
predicate pruning and takes precedence over bucket-function.type.</td>
+ </tr>
<tr>
<td><h5>bucket-function.type</h5></td>
<td style="word-wrap: break-word;">default</td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 97b91e4f29..87fc141839 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -163,6 +163,17 @@ public class CoreOptions implements Serializable {
.defaultValue(BucketFunctionType.DEFAULT)
.withDescription("The bucket function for paimon bucket.");
+ @Immutable
+ public static final ConfigOption<String> BUCKET_FUNCTION_PROVIDER =
+ key("bucket-function.provider")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional identifier of a BucketFunctionFactory
discovered from the "
+ + "classpath. When configured, the
provider is used by both "
+ + "writers and bucket predicate pruning
and takes precedence "
+ + "over bucket-function.type.");
+
/** Paimon bucket function type. */
public enum BucketFunctionType implements DescribedEnum {
DEFAULT(
@@ -3276,6 +3287,10 @@ public class CoreOptions implements Serializable {
return options.get(BUCKET_FUNCTION_TYPE);
}
+ public Optional<String> bucketFunctionProvider() {
+ return options.getOptional(BUCKET_FUNCTION_PROVIDER);
+ }
+
public Path path() {
return path(options.toMap());
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java
b/paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java
index ac3a7e7bc9..b3310d5377 100644
--- a/paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java
@@ -146,11 +146,7 @@ public class AppendOnlyFileStore extends
AbstractFileStore<InternalRow> {
public AppendOnlyFileStoreScan newScan() {
BucketSelectConverter bucketSelectConverter =
new BucketSelectConverter(
- bucketMode(),
- options.bucketFunctionType(),
- rowType,
- partitionType,
- bucketKeyType);
+ bucketMode(), options, rowType, partitionType,
bucketKeyType);
if (options().dataEvolutionEnabled()) {
return new DataEvolutionFileStoreScan(
diff --git a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
index 4b19e78eae..7675fb7525 100644
--- a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
@@ -227,7 +227,7 @@ public class KeyValueFileStore extends
AbstractFileStore<KeyValue> {
BucketSelectConverter bucketSelectConverter =
new BucketSelectConverter(
bucketMode(),
- options.bucketFunctionType(),
+ options,
schema.logicalRowType(),
partitionType,
schema.logicalBucketKeyType());
diff --git
a/paimon-core/src/main/java/org/apache/paimon/bucket/BucketFunction.java
b/paimon-core/src/main/java/org/apache/paimon/bucket/BucketFunction.java
index 6d5149dd6f..d9480a7c74 100644
--- a/paimon-core/src/main/java/org/apache/paimon/bucket/BucketFunction.java
+++ b/paimon-core/src/main/java/org/apache/paimon/bucket/BucketFunction.java
@@ -20,9 +20,11 @@ package org.apache.paimon.bucket;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.factories.FactoryUtil;
import org.apache.paimon.types.RowType;
import java.io.Serializable;
+import java.util.Optional;
/** Bucket function. */
public interface BucketFunction extends Serializable {
@@ -30,6 +32,22 @@ public interface BucketFunction extends Serializable {
int bucket(BinaryRow row, int numBuckets);
static BucketFunction create(CoreOptions options, RowType bucketKeyType) {
+ Optional<String> provider = options.bucketFunctionProvider();
+ if (provider.isPresent()) {
+ ClassLoader classLoader =
Thread.currentThread().getContextClassLoader();
+ if (classLoader == null) {
+ classLoader = BucketFunction.class.getClassLoader();
+ }
+ BucketFunction function =
+ FactoryUtil.discoverFactory(
+ classLoader, BucketFunctionFactory.class,
provider.get())
+ .create(bucketKeyType, options.toConfiguration());
+ if (function == null) {
+ throw new IllegalArgumentException(
+ "Bucket function provider '" + provider.get() + "'
returned null.");
+ }
+ return function;
+ }
CoreOptions.BucketFunctionType type = options.bucketFunctionType();
return create(type, bucketKeyType);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/bucket/BucketFunctionFactory.java
b/paimon-core/src/main/java/org/apache/paimon/bucket/BucketFunctionFactory.java
new file mode 100644
index 0000000000..a48b1ece9e
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/bucket/BucketFunctionFactory.java
@@ -0,0 +1,45 @@
+/*
+ * 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.bucket;
+
+import org.apache.paimon.factories.Factory;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.RowType;
+
+/**
+ * Factory for a bucket function shared by table writers and predicate pruning.
+ *
+ * <p>The identifier is persisted as part of the table's physical layout.
Implementations must
+ * produce the same bucket for the same logical key and bucket count across
processes and releases.
+ * Incompatible algorithm changes require a new identifier.
+ */
+public interface BucketFunctionFactory extends Factory {
+
+ /**
+ * Creates the bucket function for a table.
+ *
+ * <p>The function must not retain or modify the supplied key row and must
return a bucket in
+ * {@code [0, numBuckets)}. A factory can be cached by the runtime; it
must create functions
+ * without relying on per-call mutable factory state. Returned functions
are not assumed to be
+ * safe to share between concurrent callers.
+ *
+ * @return a non-null bucket function
+ */
+ BucketFunction create(RowType bucketKeyType, Options tableOptions);
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/BucketSelectConverter.java
b/paimon-core/src/main/java/org/apache/paimon/operation/BucketSelectConverter.java
index e441641f7e..2989107383 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/BucketSelectConverter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/BucketSelectConverter.java
@@ -18,6 +18,7 @@
package org.apache.paimon.operation;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.CoreOptions.BucketFunctionType;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.predicate.Predicate;
@@ -25,6 +26,7 @@ import org.apache.paimon.table.BucketMode;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.TriFilter;
+import java.util.Collections;
import java.util.Optional;
import java.util.Set;
@@ -34,19 +36,41 @@ import static
org.apache.paimon.predicate.PredicateVisitor.collectFieldNames;
public class BucketSelectConverter {
private final BucketMode bucketMode;
- private final BucketFunctionType bucketFunctionType;
+ private final CoreOptions coreOptions;
private final RowType rowType;
private final RowType partitionType;
private final RowType bucketKeyType;
+ /**
+ * @deprecated Use {@link #BucketSelectConverter(BucketMode, CoreOptions,
RowType, RowType,
+ * RowType)} so custom bucket function providers are preserved.
+ */
+ @Deprecated
public BucketSelectConverter(
BucketMode bucketMode,
BucketFunctionType bucketFunctionType,
RowType rowType,
RowType partitionType,
RowType bucketKeyType) {
+ this(
+ bucketMode,
+ new CoreOptions(
+ Collections.singletonMap(
+ CoreOptions.BUCKET_FUNCTION_TYPE.key(),
+ bucketFunctionType.toString())),
+ rowType,
+ partitionType,
+ bucketKeyType);
+ }
+
+ public BucketSelectConverter(
+ BucketMode bucketMode,
+ CoreOptions coreOptions,
+ RowType rowType,
+ RowType partitionType,
+ RowType bucketKeyType) {
this.bucketMode = bucketMode;
- this.bucketFunctionType = bucketFunctionType;
+ this.coreOptions = coreOptions;
this.rowType = rowType;
this.partitionType = partitionType;
this.bucketKeyType = bucketKeyType;
@@ -67,7 +91,6 @@ public class BucketSelectConverter {
}
return Optional.of(
- new BucketSelector(
- predicate, bucketFunctionType, rowType, partitionType,
bucketKeyType));
+ new BucketSelector(predicate, coreOptions, rowType,
partitionType, bucketKeyType));
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/BucketSelector.java
b/paimon-core/src/main/java/org/apache/paimon/operation/BucketSelector.java
index 0eeb0b314e..7f5592cccd 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/BucketSelector.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/BucketSelector.java
@@ -18,6 +18,7 @@
package org.apache.paimon.operation;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.CoreOptions.BucketFunctionType;
import org.apache.paimon.bucket.BucketFunction;
import org.apache.paimon.data.BinaryRow;
@@ -38,6 +39,7 @@ import
org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableSet;
import javax.annotation.concurrent.ThreadSafe;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -59,7 +61,7 @@ public class BucketSelector implements ManifestBucketFilter {
public static final int MAX_VALUES = 1000;
- private final BucketFunctionType bucketFunctionType;
+ private final CoreOptions coreOptions;
private final RowType rowType;
private final RowType partitionType;
private final RowType bucketKeyType;
@@ -67,14 +69,36 @@ public class BucketSelector implements ManifestBucketFilter
{
private final Map<BinaryRow, Optional<PartitionSelector>>
partitionSelectors;
private final Optional<PartitionSelector> manifestSelector;
+ /**
+ * @deprecated Use {@link #BucketSelector(Predicate, CoreOptions, RowType,
RowType, RowType)} so
+ * custom bucket function providers are preserved.
+ */
+ @Deprecated
public BucketSelector(
Predicate predicate,
BucketFunctionType bucketFunctionType,
RowType rowType,
RowType partitionType,
RowType bucketKeyType) {
+ this(
+ predicate,
+ new CoreOptions(
+ Collections.singletonMap(
+ CoreOptions.BUCKET_FUNCTION_TYPE.key(),
+ bucketFunctionType.toString())),
+ rowType,
+ partitionType,
+ bucketKeyType);
+ }
+
+ public BucketSelector(
+ Predicate predicate,
+ CoreOptions coreOptions,
+ RowType rowType,
+ RowType partitionType,
+ RowType bucketKeyType) {
this.predicate = predicate;
- this.bucketFunctionType = bucketFunctionType;
+ this.coreOptions = coreOptions;
this.rowType = rowType;
this.partitionType = partitionType;
this.bucketKeyType = bucketKeyType;
@@ -136,7 +160,9 @@ public class BucketSelector implements ManifestBucketFilter
{
@SuppressWarnings("unchecked")
List<Object>[] bucketValues = new List[bucketKeyType.getFieldCount()];
- BucketFunction bucketFunction =
BucketFunction.create(bucketFunctionType, bucketKeyType);
+ // Factory discovery is cached by ClassLoader; functions need not be
safe to share
+ // across concurrent partition selectors.
+ BucketFunction bucketFunction = BucketFunction.create(coreOptions,
bucketKeyType);
nextAnd:
for (Predicate andPredicate : splitAnd(bucketPredicate)) {
@@ -243,7 +269,9 @@ public class BucketSelector implements ManifestBucketFilter
{
.contains(bucket);
}
- private Set<Integer> createBucketSet(int numBucket) {
+ // Different bucket counts can miss the cache concurrently. The
provider function is
+ // confined to this selector, but is not required to support
overlapping calls.
+ private synchronized Set<Integer> createBucketSet(int numBucket) {
ImmutableSet.Builder<Integer> builder = new
ImmutableSet.Builder<>();
for (BinaryRow key : bucketKeys) {
builder.add(bucketFunction.bucket(key, numBucket));
diff --git
a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index c98600f7ee..e279d64795 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -23,6 +23,7 @@ import org.apache.paimon.CoreOptions.ChangelogProducer;
import org.apache.paimon.CoreOptions.MapStorageLayout;
import org.apache.paimon.CoreOptions.MergeEngine;
import org.apache.paimon.TableType;
+import org.apache.paimon.bucket.BucketFunction;
import org.apache.paimon.data.shredding.MapSharedShreddingUtils;
import org.apache.paimon.factories.FactoryUtil;
import org.apache.paimon.fileindex.FileIndexOptions;
@@ -1521,6 +1522,13 @@ public class SchemaValidation {
private static void validateBucket(TableSchema schema, CoreOptions
options) {
int bucket = options.bucket();
+ if (options.bucketFunctionProvider().isPresent()) {
+ checkArgument(
+ bucket > 0,
+ "A custom bucket-function.provider requires a fixed
positive bucket count.");
+ // Validate provider discovery and bucket-key compatibility before
creating the table.
+ BucketFunction.create(options, schema.logicalBucketKeyType());
+ }
if (bucket == -1) {
if (options.toMap().get(BUCKET_KEY.key()) != null) {
throw new IllegalArgumentException(
diff --git
a/paimon-core/src/test/java/org/apache/paimon/bucket/BucketFunctionProviderTest.java
b/paimon-core/src/test/java/org/apache/paimon/bucket/BucketFunctionProviderTest.java
new file mode 100644
index 0000000000..98279d7b4b
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/bucket/BucketFunctionProviderTest.java
@@ -0,0 +1,46 @@
+/*
+ * 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.bucket;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.factories.FactoryException;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for bucket function provider discovery. */
+class BucketFunctionProviderTest {
+
+ @Test
+ void missingProviderDoesNotFallBackToDefault() {
+ CoreOptions options =
+ new CoreOptions(
+ Collections.singletonMap(
+ "bucket-function.provider",
"missing-test-provider"));
+ RowType keyType = DataTypes.ROW(DataTypes.FIELD(0, "id",
DataTypes.INT()));
+
+ assertThatThrownBy(() -> BucketFunction.create(options, keyType))
+ .isInstanceOf(FactoryException.class);
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/bucket/ConcurrentBucketFunctionFactory.java
b/paimon-core/src/test/java/org/apache/paimon/bucket/ConcurrentBucketFunctionFactory.java
new file mode 100644
index 0000000000..9f31da22b9
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/bucket/ConcurrentBucketFunctionFactory.java
@@ -0,0 +1,67 @@
+/*
+ * 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.bucket;
+
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.RowType;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/** A deterministic bucket provider which detects concurrent use of a function
instance. */
+public class ConcurrentBucketFunctionFactory implements BucketFunctionFactory {
+
+ public static final String IDENTIFIER = "test-concurrent-bucket";
+ public static final String CHECK_CONCURRENCY =
"test.bucket-function.check-concurrency";
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public BucketFunction create(RowType bucketKeyType, Options tableOptions) {
+ if (!"true".equals(tableOptions.toMap().get(CHECK_CONCURRENCY))) {
+ return (row, numBuckets) -> numBuckets - 1;
+ }
+ AtomicBoolean active = new AtomicBoolean();
+ AtomicBoolean firstCall = new AtomicBoolean(true);
+ CountDownLatch overlap = new CountDownLatch(1);
+ return (row, numBuckets) -> {
+ if (!active.compareAndSet(false, true)) {
+ overlap.countDown();
+ throw new IllegalStateException("Concurrent bucket function
invocation");
+ }
+ try {
+ if (firstCall.getAndSet(false)) {
+ // Give another manifest worker time to request a
different bucket count.
+ // Correctly serialized callers cannot enter until this
bounded wait ends.
+ overlap.await(1, TimeUnit.SECONDS);
+ }
+ return numBuckets - 1;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ } finally {
+ active.set(false);
+ }
+ };
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/bucket/TestBucketFunctionFactory.java
b/paimon-core/src/test/java/org/apache/paimon/bucket/TestBucketFunctionFactory.java
new file mode 100644
index 0000000000..f3d03d933b
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/bucket/TestBucketFunctionFactory.java
@@ -0,0 +1,38 @@
+/*
+ * 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.bucket;
+
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.RowType;
+
+/** Bucket function factory used to verify provider discovery in tests. */
+public class TestBucketFunctionFactory implements BucketFunctionFactory {
+
+ public static final String IDENTIFIER = "test-last-bucket";
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public BucketFunction create(RowType bucketKeyType, Options tableOptions) {
+ return (row, numBuckets) -> numBuckets - 1;
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/BucketSelectorTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/BucketSelectorTest.java
index 16da67c541..e7762f52e6 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/BucketSelectorTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/BucketSelectorTest.java
@@ -18,7 +18,9 @@
package org.apache.paimon.operation;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.CoreOptions.BucketFunctionType;
+import org.apache.paimon.bucket.TestBucketFunctionFactory;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.manifest.BucketFilter;
import org.apache.paimon.manifest.ManifestFileMeta;
@@ -30,6 +32,7 @@ import org.apache.paimon.types.RowType;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
@@ -60,6 +63,26 @@ public class BucketSelectorTest {
assertThat(selected).hasSize(1);
}
+ @Test
+ public void testCustomProviderIsUsedByPredicatePruning() {
+ RowType rowType = DataTypes.ROW(DataTypes.FIELD(0, "k",
DataTypes.INT()));
+ RowType partType = RowType.of();
+ Predicate predicate = new PredicateBuilder(rowType).equal(0, 5);
+ CoreOptions options =
+ new CoreOptions(
+ Collections.singletonMap(
+ CoreOptions.BUCKET_FUNCTION_PROVIDER.key(),
+ TestBucketFunctionFactory.IDENTIFIER));
+
+ BucketSelector selector =
+ new BucketSelector(predicate, options, rowType, partType,
rowType);
+
+ assertThat(selectedBuckets(selector, BinaryRow.EMPTY_ROW, NUM_BUCKETS))
+ .containsExactly(NUM_BUCKETS - 1);
+ assertThat(selector.mayContain(0, NUM_BUCKETS - 2,
NUM_BUCKETS)).isFalse();
+ assertThat(selector.mayContain(NUM_BUCKETS - 1, NUM_BUCKETS - 1,
NUM_BUCKETS)).isTrue();
+ }
+
@Test
public void testPostponeBucketIsNeverPruned() {
RowType rowType = DataTypes.ROW(DataTypes.FIELD(0, "k",
DataTypes.INT()));
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/BucketFilterScanTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/BucketFilterScanTest.java
index 94494e0406..b9031c13c3 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/BucketFilterScanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/BucketFilterScanTest.java
@@ -18,16 +18,25 @@
package org.apache.paimon.table;
+import org.apache.paimon.bucket.ConcurrentBucketFunctionFactory;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.types.DataTypes;
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
@@ -61,6 +70,53 @@ public class BucketFilterScanTest extends TableTestBase {
testCompositeBucketFilterWithCompoundPredicate(true);
}
+ @Test
+ public void testProviderIsConfinedWhenScanningOverwriteHistory() throws
Exception {
+ Identifier tableId = identifier("provider_overwrite_history");
+ catalog.createTable(
+ tableId,
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .option(BUCKET.key(), "2")
+ .option(BUCKET_KEY.key(), "id")
+ .option(
+ "bucket-function.provider",
+ ConcurrentBucketFunctionFactory.IDENTIFIER)
+ .option("manifest.merge-min-count", "100")
+ .option("scan.manifest.parallelism", "4")
+ .build(),
+ false);
+
+ // Committed DELETE manifests retain the old bucket counts during a
filtered scan.
+ for (int buckets = 2; buckets <= 4; buckets++) {
+ catalog.alterTable(
+ tableId,
+
Collections.singletonList(SchemaChange.setOption(BUCKET.key(), "" + buckets)),
+ false);
+ Table table = catalog.getTable(tableId);
+ BatchWriteBuilder builder =
table.newBatchWriteBuilder().withOverwrite();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ write.write(GenericRow.of(17));
+ commit.commit(write.prepareCommit());
+ }
+ }
+
+ Table table =
+ catalog.getTable(tableId)
+ .copy(
+ Collections.singletonMap(
+
ConcurrentBucketFunctionFactory.CHECK_CONCURRENCY, "true"));
+ ReadBuilder read =
+ table.newReadBuilder()
+ .withFilter(new
PredicateBuilder(table.rowType()).equal(0, 17));
+ List<Integer> ids = new ArrayList<>();
+ read.newRead()
+ .createReader(read.newScan().plan())
+ .forEachRemaining(row -> ids.add(row.getInt(0)));
+ assertThat(ids).containsExactly(17);
+ }
+
/**
* Tests bucket filtering with compound predicates on a single-field
bucket key.
*
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/sink/FixedBucketRowKeyExtractorTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/sink/FixedBucketRowKeyExtractorTest.java
index 5c551dac70..39d69dd31a 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/sink/FixedBucketRowKeyExtractorTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/sink/FixedBucketRowKeyExtractorTest.java
@@ -19,6 +19,7 @@
package org.apache.paimon.table.sink;
import org.apache.paimon.bucket.DefaultBucketFunction;
+import org.apache.paimon.bucket.TestBucketFunctionFactory;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
@@ -41,6 +42,7 @@ import java.util.Map;
import java.util.concurrent.ThreadLocalRandom;
import static org.apache.paimon.CoreOptions.BUCKET;
+import static org.apache.paimon.CoreOptions.BUCKET_FUNCTION_PROVIDER;
import static org.apache.paimon.CoreOptions.BUCKET_KEY;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -71,6 +73,33 @@ public class FixedBucketRowKeyExtractorTest {
assertThat(bucket(extractor("", "a,b,c"), row)).isEqualTo(40);
}
+ @Test
+ public void testCustomBucketFunctionProvider() {
+ RowType rowType =
+ new RowType(
+ Arrays.asList(
+ new DataField(0, "a", new IntType()),
+ new DataField(1, "b", new IntType()),
+ new DataField(2, "c", new IntType())));
+ List<DataField> fields = TableSchema.newFields(rowType);
+ Map<String, String> options = new HashMap<>();
+ options.put(BUCKET_KEY.key(), "a");
+ options.put(BUCKET.key(), "100");
+ options.put(BUCKET_FUNCTION_PROVIDER.key(),
TestBucketFunctionFactory.IDENTIFIER);
+ FixedBucketRowKeyExtractor extractor =
+ new FixedBucketRowKeyExtractor(
+ new TableSchema(
+ 0,
+ fields,
+ RowType.currentHighestFieldId(fields),
+ Collections.emptyList(),
+ Collections.singletonList("a"),
+ options,
+ ""));
+
+ assertThat(bucket(extractor, GenericRow.of(5, 6, 7))).isEqualTo(99);
+ }
+
@Test
public void testIllegalBucket() {
GenericRow row = GenericRow.of(5, 6, 7);
diff --git
a/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
b/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
index f22e050a9d..22ea0ab4fc 100644
---
a/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
+++
b/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -16,6 +16,8 @@
org.apache.paimon.mergetree.compact.aggregate.TestCustomAggFactory
org.apache.paimon.mergetree.compact.aggregate.TestMapOnlyAggFactory
org.apache.paimon.rest.auth.CustomTestDLFTokenLoaderFactory
+org.apache.paimon.bucket.TestBucketFunctionFactory
+org.apache.paimon.bucket.ConcurrentBucketFunctionFactory
org.apache.paimon.iceberg.RecordingIcebergMetadataCommitter$Factory
org.apache.paimon.catalog.TestCatalogLockFactory
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 bb674f3dd4..b3f19e73f7 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
@@ -18,6 +18,7 @@
package org.apache.paimon.spark
+import org.apache.paimon.CoreOptions.BucketFunctionType
import org.apache.paimon.partition.PartitionPredicate
import org.apache.paimon.predicate.{FullTextSearch, HybridSearch, Predicate,
TopN, VectorSearch}
import org.apache.paimon.spark.read.{BinPackingSplits, VariantExtractionInfo}
@@ -51,7 +52,11 @@ case class PaimonScan(
table match {
case fileStoreTable: FileStoreTable =>
val bucketSpec = fileStoreTable.bucketSpec()
- if (bucketSpec.getBucketMode != BucketMode.HASH_FIXED) {
+ if (
+ bucketSpec.getBucketMode != BucketMode.HASH_FIXED ||
+ coreOptions.bucketFunctionProvider().isPresent ||
+ coreOptions.bucketFunctionType() != BucketFunctionType.DEFAULT
+ ) {
None
} else if (bucketSpec.getBucketKeys.size() > 1) {
None
diff --git
a/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/CustomBucketFunction33Test.scala
b/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/CustomBucketFunction33Test.scala
new file mode 100644
index 0000000000..d6bb61935d
--- /dev/null
+++
b/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/CustomBucketFunction33Test.scala
@@ -0,0 +1,21 @@
+/*
+ * 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.sql
+
+class CustomBucketFunction33Test extends CustomBucketFunctionTest
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 96d65bbcb7..24895b3821 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
@@ -61,7 +61,8 @@ case class PaimonScan(
val bucketSpec = fileStoreTable.bucketSpec()
// todo introduce bucket transform for different bucket function type
if (
- bucketSpec.getBucketMode != BucketMode.HASH_FIXED || coreOptions
+ bucketSpec.getBucketMode != BucketMode.HASH_FIXED ||
+ coreOptions.bucketFunctionProvider().isPresent || coreOptions
.bucketFunctionType() != BucketFunctionType.DEFAULT
) {
None
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkTableBase.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkTableBase.scala
index d139ed025b..968e761d78 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkTableBase.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkTableBase.scala
@@ -54,6 +54,7 @@ abstract class PaimonSparkTableBase(val table: Table)
}
private def supportsV2Write: Boolean = {
+ !coreOptions.bucketFunctionProvider().isPresent &&
coreOptions.bucketFunctionType() == BucketFunctionType.DEFAULT && {
table match {
case storeTable: FileStoreTable =>
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
index 132b8bfe47..605284ec2e 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
@@ -356,7 +356,10 @@ case class PaimonSparkWriter(
writeWithoutBucket(input)
case HASH_FIXED =>
- if (paimonExtensionEnabled(sparkSession) &&
BucketFunction.supportsTable(table)) {
+ if (
+ paimonExtensionEnabled(sparkSession) &&
+ !coreOptions.bucketFunctionProvider().isPresent &&
BucketFunction.supportsTable(table)
+ ) {
// Topology: input -> shuffle by partition & bucket
val bucketNumber = coreOptions.bucket()
val bucketKeyCol = tableSchema
diff --git
a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/ShiftedBucketFunctionFactory.java
b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/ShiftedBucketFunctionFactory.java
new file mode 100644
index 0000000000..863f2aaf90
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/ShiftedBucketFunctionFactory.java
@@ -0,0 +1,43 @@
+/*
+ * 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.bucket.BucketFunction;
+import org.apache.paimon.bucket.BucketFunctionFactory;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.RowType;
+
+import java.util.Collections;
+
+/** A schema-sensitive, non-default bucket function for Spark integration
tests. */
+public class ShiftedBucketFunctionFactory implements BucketFunctionFactory {
+
+ @Override
+ public String identifier() {
+ return "spark-test-shifted";
+ }
+
+ @Override
+ public BucketFunction create(RowType bucketKeyType, Options tableOptions) {
+ if
(!bucketKeyType.getFieldNames().equals(Collections.singletonList("id"))) {
+ throw new IllegalArgumentException("Expected the schema's id
bucket key");
+ }
+ return (row, numBuckets) -> Math.floorMod(row.getInt(0) + 1,
numBuckets);
+ }
+}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
b/paimon-spark/paimon-spark-ut/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
new file mode 100644
index 0000000000..626d6a62f4
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -0,0 +1,18 @@
+# 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.
+
+org.apache.paimon.spark.ShiftedBucketFunctionFactory
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CustomBucketFunctionTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CustomBucketFunctionTest.scala
new file mode 100644
index 0000000000..5c18e93176
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CustomBucketFunctionTest.scala
@@ -0,0 +1,125 @@
+/*
+ * 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.sql
+
+import org.apache.paimon.CoreOptions.BucketFunctionType
+import org.apache.paimon.spark.{PaimonSparkTestBase, SparkTable}
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.connector.catalog.TableCapability
+import
org.apache.spark.sql.connector.read.partitioning.{KeyGroupedPartitioning,
UnknownPartitioning}
+import org.apache.spark.sql.execution.CommandResultExec
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.execution.datasources.v2.AppendDataExecV1
+
+class CustomBucketFunctionTest extends PaimonSparkTestBase with
AdaptiveSparkPlanHelper {
+
+ private def createTable(name: String, primaryKey: Boolean, custom: Boolean =
true): Unit = {
+ val key = if (primaryKey) "'primary-key' = 'pt,id'," else ""
+ val provider = if (custom) {
+ "'bucket-function.provider' = 'spark-test-shifted',"
+ } else {
+ ""
+ }
+ sql(s"""CREATE TABLE $name (pt STRING, payload STRING, id INT)
+ |PARTITIONED BY (pt) TBLPROPERTIES (
+ |$key $provider 'bucket' = '4', 'bucket-key' = 'id')""".stripMargin)
+ }
+
+ test("custom provider does not advertise V2 write with the default bucket
enum") {
+ withSparkSQLConf("spark.paimon.write.use-v2-write" -> "true") {
+ withTable("custom_bucket", "default_bucket") {
+ createTable("custom_bucket", primaryKey = true)
+ createTable("default_bucket", primaryKey = true, custom = false)
+ val table = loadTable("custom_bucket")
+ assert(table.coreOptions().bucketFunctionType() ==
BucketFunctionType.DEFAULT)
+ assert(SparkTable(loadTable("default_bucket")).useV2Write ==
gteqSpark3_4)
+ assert(!SparkTable(table).useV2Write)
+
assert(!SparkTable(table).capabilities().contains(TableCapability.BATCH_WRITE))
+ }
+ }
+ }
+
+ for (primaryKey <- Seq(false, true); useV2 <- Seq(false, true)) {
+ test(s"custom provider assigns persisted buckets: primaryKey=$primaryKey,
useV2=$useV2") {
+ withSparkSQLConf("spark.paimon.write.use-v2-write" -> useV2.toString) {
+ withTable("custom_bucket") {
+ createTable("custom_bucket", primaryKey)
+ val inserted = sql("""INSERT INTO custom_bucket VALUES
+ |('p', 'a', 0), ('p', 'b', 1),
+ |('q', 'c', 2), ('q', 'd', 3)""".stripMargin)
+ val plan = inserted.queryExecution.executedPlan
+ .asInstanceOf[CommandResultExec]
+ .commandPhysicalPlan
+ assert(collect(plan) { case append: AppendDataExecV1 => append
}.size == 1)
+ // The provider maps id 0,1,2,3 to buckets 1,2,3,0, respectively.
+ checkAnswer(
+ sql("SELECT id, __paimon_bucket FROM custom_bucket"),
+ Seq(Row(0, 1), Row(1, 2), Row(2, 3), Row(3, 0)))
+ // Equality pruning must agree with the writer's provider-based
assignment.
+ checkAnswer(sql("SELECT payload FROM custom_bucket WHERE id = 0"),
Row("a"))
+ }
+ }
+ }
+ }
+
+ test("custom provider scan does not advertise default bucket partitioning") {
+ assume(gteqSpark3_3)
+ withSparkSQLConf(
+ "spark.paimon.write.use-v2-write" -> "false",
+ "spark.paimon.scan.preserve-data-grouping" -> "true",
+ "spark.sql.sources.v2.bucketing.enabled" -> "true") {
+ withTable("custom_bucket", "default_bucket") {
+ createTable("custom_bucket", primaryKey = true)
+ createTable("default_bucket", primaryKey = true, custom = false)
+ Seq("custom_bucket", "default_bucket").foreach {
+ name => sql(s"INSERT INTO $name VALUES ('p', 'a', 0), ('p', 'b', 1)")
+ }
+ assert(
+ getPaimonScan("SELECT * FROM default_bucket").outputPartitioning
+ .isInstanceOf[KeyGroupedPartitioning])
+ val scan = getPaimonScan("SELECT * FROM custom_bucket")
+ assert(scan.outputPartitioning.isInstanceOf[UnknownPartitioning])
+ if (gteqSpark3_4) {
+ assert(scan.outputOrdering().isEmpty)
+ }
+ }
+ }
+ }
+
+ for (bucketFunction <- Seq("mod", "hive")) {
+ test(s"$bucketFunction scan does not advertise default bucket
partitioning") {
+ assume(gteqSpark3_3)
+ withSparkSQLConf(
+ "spark.paimon.write.use-v2-write" -> "false",
+ "spark.paimon.scan.preserve-data-grouping" -> "true",
+ "spark.sql.sources.v2.bucketing.enabled" -> "true") {
+ withTable("typed_bucket") {
+ sql(s"""CREATE TABLE typed_bucket (id INT, payload STRING)
TBLPROPERTIES (
+ |'primary-key' = 'id', 'bucket' = '4',
+ |'bucket-function.type' = '$bucketFunction')""".stripMargin)
+ sql("INSERT INTO typed_bucket VALUES (0, 'a'), (1, 'b'), (2, 'c'),
(3, 'd')")
+ val scan = getPaimonScan("SELECT * FROM typed_bucket")
+ assert(scan.outputPartitioning.isInstanceOf[UnknownPartitioning])
+ checkAnswer(sql("SELECT payload FROM typed_bucket WHERE id = 1"),
Row("b"))
+ }
+ }
+ }
+ }
+}
