From a574f6dbcfacbedb45d21490dcf7381b85cc8b91 Mon Sep 17 00:00:00 2001 From: fhan Date: Mon, 26 Oct 2026 20:33:40 +0800 Subject: [PATCH] [flink] Add $partitions virtual table --- .../Flink120PartitionsPlanCacheITCase.java | 192 +++++++++++ .../fluss/flink/catalog/FlinkCatalog.java | 64 +++- .../flink/catalog/FlinkTableFactory.java | 19 ++ .../ListPartitionInfosProcedure.java | 4 + .../fluss/flink/source/BucketsSource.java | 294 +--------------- .../source/PartitionsFlinkTableSource.java | 67 ++++ .../fluss/flink/source/PartitionsSource.java | 84 +++++ .../source/SingleSplitMetadataSource.java | 323 ++++++++++++++++++ .../flink/catalog/FlinkCatalogITCase.java | 83 +++++ .../flink/catalog/FlinkTableFactoryTest.java | 45 +++ .../source/BucketsFlinkTableSourceTest.java | 16 +- .../PartitionsFlinkTableSourceTest.java | 89 +++++ website/docs/engine-flink/procedures.md | 12 + .../data-distribution/bucketing.md | 5 +- website/docs/table-design/virtual-tables.md | 40 ++- 15 files changed, 1034 insertions(+), 303 deletions(-) create mode 100644 fluss-flink/fluss-flink-1.20/src/test/java/org/apache/fluss/flink/catalog/Flink120PartitionsPlanCacheITCase.java create mode 100644 fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/PartitionsFlinkTableSource.java create mode 100644 fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/PartitionsSource.java create mode 100644 fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/SingleSplitMetadataSource.java create mode 100644 fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/PartitionsFlinkTableSourceTest.java diff --git a/fluss-flink/fluss-flink-1.20/src/test/java/org/apache/fluss/flink/catalog/Flink120PartitionsPlanCacheITCase.java b/fluss-flink/fluss-flink-1.20/src/test/java/org/apache/fluss/flink/catalog/Flink120PartitionsPlanCacheITCase.java new file mode 100644 index 00000000000..f697a152829 --- /dev/null +++ b/fluss-flink/fluss-flink-1.20/src/test/java/org/apache/fluss/flink/catalog/Flink120PartitionsPlanCacheITCase.java @@ -0,0 +1,192 @@ +/* + * 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.fluss.flink.catalog; + +import org.apache.fluss.client.Connection; +import org.apache.fluss.client.ConnectionFactory; +import org.apache.fluss.client.admin.Admin; +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.metadata.PartitionSpec; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.server.testutils.FlussClusterExtension; + +import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.gateway.api.endpoint.EndpointVersion; +import org.apache.flink.table.gateway.api.operation.OperationHandle; +import org.apache.flink.table.gateway.api.session.SessionEnvironment; +import org.apache.flink.table.gateway.api.session.SessionHandle; +import org.apache.flink.table.gateway.service.SqlGatewayServiceImpl; +import org.apache.flink.table.gateway.service.utils.SqlGatewayServiceExtension; +import org.apache.flink.test.junit5.MiniClusterExtension; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Order; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import java.time.Duration; +import java.util.Collections; +import java.util.List; + +import static org.apache.flink.table.gateway.api.config.SqlGatewayServiceConfigOptions.SQL_GATEWAY_SESSION_PLAN_CACHE_ENABLED; +import static org.apache.flink.table.gateway.service.utils.SqlGatewayServiceTestUtil.awaitOperationTermination; +import static org.apache.flink.table.gateway.service.utils.SqlGatewayServiceTestUtil.fetchAllResults; +import static org.apache.fluss.testutils.common.CommonTestUtils.retry; +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests that cached SQL Gateway plans load fresh {@code $partitions} metadata at runtime. */ +public class Flink120PartitionsPlanCacheITCase { + + private static final String CATALOG_NAME = "testcatalog"; + private static final String DEFAULT_DB = FlinkCatalogOptions.DEFAULT_DATABASE.defaultValue(); + + @RegisterExtension + public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION = + FlussClusterExtension.builder().setNumOfTabletServers(1).build(); + + @RegisterExtension + @Order(1) + static final MiniClusterExtension MINI_CLUSTER = + new MiniClusterExtension( + new MiniClusterResourceConfiguration.Builder() + .setNumberTaskManagers(1) + .build()); + + @RegisterExtension + @Order(2) + static final SqlGatewayServiceExtension SQL_GATEWAY_SERVICE_EXTENSION = + new SqlGatewayServiceExtension(MINI_CLUSTER::getClientConfiguration); + + private static SqlGatewayServiceImpl service; + + private SessionHandle sessionHandle; + + @BeforeAll + static void beforeAll() { + service = (SqlGatewayServiceImpl) SQL_GATEWAY_SERVICE_EXTENSION.getService(); + } + + @BeforeEach + void before() { + SessionEnvironment sessionEnvironment = + SessionEnvironment.newBuilder() + .setSessionEndpointVersion(new EndpointVersion() {}) + .addSessionConfig( + Collections.singletonMap( + SQL_GATEWAY_SESSION_PLAN_CACHE_ENABLED.key(), "true")) + .build(); + sessionHandle = service.openSession(sessionEnvironment); + + Configuration flussConf = FLUSS_CLUSTER_EXTENSION.getClientConfig(); + String bootstrapServers = String.join(",", flussConf.get(ConfigOptions.BOOTSTRAP_SERVERS)); + service.configureSession( + sessionHandle, + String.format( + "CREATE CATALOG %s WITH ('type' = 'fluss', 'bootstrap.servers' = '%s')", + CATALOG_NAME, bootstrapServers), + -1); + service.configureSession(sessionHandle, "USE CATALOG " + CATALOG_NAME, -1); + } + + @AfterEach + void after() { + if (sessionHandle != null) { + service.closeSession(sessionHandle); + } + } + + @Test + void testCachedPlanLoadsFreshPartitionMetadata() throws Exception { + String tableName = "cached_plan_partitions"; + TablePath tablePath = TablePath.of(DEFAULT_DB, tableName); + executeUpdate( + "CREATE TABLE " + + tableName + + " (id INT, dt STRING) PARTITIONED BY (dt) " + + "WITH ('bucket.num' = '2')"); + + String query = + "SELECT partition_name, bucket_count FROM " + + tableName + + FlinkCatalog.PARTITIONS_TABLE_SUFFIX; + + try (Connection connection = + ConnectionFactory.createConnection( + FLUSS_CLUSTER_EXTENSION.getClientConfig()); + Admin admin = connection.getAdmin()) { + admin.createPartition( + tablePath, + new PartitionSpec(Collections.singletonMap("dt", "2026-10-25")), + false) + .get(); + waitForPartitionCount(admin, tablePath, 1); + + List initialRows = executeQuery(query); + assertThat(initialRows).hasSize(1); + assertThat(initialRows.get(0).getString(0).toString()).isEqualTo("2026-10-25"); + assertThat(initialRows.get(0).getInt(1)).isEqualTo(2); + + admin.createPartition( + tablePath, + new PartitionSpec(Collections.singletonMap("dt", "2026-10-26")), + false) + .get(); + waitForPartitionCount(admin, tablePath, 2); + + List refreshedRows = executeQuery(query); + assertThat(refreshedRows).hasSize(2); + assertThat(refreshedRows) + .extracting(row -> row.getString(0).toString()) + .containsExactlyInAnyOrder("2026-10-25", "2026-10-26"); + assertThat(refreshedRows).extracting(row -> row.getInt(1)).containsOnly(2); + } + } + + private void executeUpdate(String statement) throws Exception { + OperationHandle operationHandle = + service.executeStatement( + sessionHandle, + statement, + -1, + new org.apache.flink.configuration.Configuration()); + awaitOperationTermination(service, sessionHandle, operationHandle); + service.closeOperation(sessionHandle, operationHandle); + } + + private List executeQuery(String statement) throws Exception { + OperationHandle operationHandle = + service.executeStatement( + sessionHandle, + statement, + -1, + new org.apache.flink.configuration.Configuration()); + awaitOperationTermination(service, sessionHandle, operationHandle); + List results = fetchAllResults(service, sessionHandle, operationHandle); + service.closeOperation(sessionHandle, operationHandle); + return results; + } + + private void waitForPartitionCount(Admin admin, TablePath tablePath, int expectedCount) { + retry( + Duration.ofMinutes(1), + () -> assertThat(admin.listPartitionInfos(tablePath).get()).hasSize(expectedCount)); + } +} diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java index 31f56ddc134..95bb01cc2de 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java @@ -142,6 +142,7 @@ public class FlinkCatalog extends AbstractCatalog { public static final String CHANGELOG_TABLE_SUFFIX = "$changelog"; public static final String BINLOG_TABLE_SUFFIX = "$binlog"; public static final String BUCKETS_TABLE_SUFFIX = "$buckets"; + public static final String PARTITIONS_TABLE_SUFFIX = "$partitions"; protected final ClassLoader classLoader; @@ -446,6 +447,9 @@ public CatalogBaseTable getTable(ObjectPath objectPath) } else if (tableName.endsWith(BUCKETS_TABLE_SUFFIX) && !tableName.contains(LAKE_TABLE_SPLITTER)) { return getVirtualBucketsTable(objectPath); + } else if (tableName.endsWith(PARTITIONS_TABLE_SUFFIX) + && !tableName.contains(LAKE_TABLE_SPLITTER)) { + return getVirtualPartitionsTable(objectPath); } TablePath tablePath = toTablePath(objectPath); @@ -979,8 +983,8 @@ protected TablePath toTablePath(ObjectPath objectPath) { /** * Converts an {@link ObjectPath} to a physical {@link TablePath}, stripping any virtual table - * suffix ({@code $changelog}, {@code $binlog}, or {@code $buckets}) if present. This is needed - * because virtual tables are resolved from their base table. + * suffix ({@code $changelog}, {@code $binlog}, {@code $buckets}, or {@code $partitions}) if + * present. This is needed because virtual tables are resolved from their base table. */ private TablePath toPhysicalTablePath(ObjectPath objectPath) { String tableName = objectPath.getObjectName(); @@ -994,6 +998,9 @@ private TablePath toPhysicalTablePath(ObjectPath objectPath) { tableName = tableName.substring(0, tableName.length() - BINLOG_TABLE_SUFFIX.length()); } else if (tableName.endsWith(BUCKETS_TABLE_SUFFIX)) { tableName = tableName.substring(0, tableName.length() - BUCKETS_TABLE_SUFFIX.length()); + } else if (tableName.endsWith(PARTITIONS_TABLE_SUFFIX)) { + tableName = + tableName.substring(0, tableName.length() - PARTITIONS_TABLE_SUFFIX.length()); } return TablePath.of(objectPath.getDatabaseName(), tableName); } @@ -1007,7 +1014,8 @@ static boolean isVirtualTableName(String tableName) { return !tableName.contains(LAKE_TABLE_SPLITTER) && (tableName.endsWith(CHANGELOG_TABLE_SUFFIX) || tableName.endsWith(BINLOG_TABLE_SUFFIX) - || tableName.endsWith(BUCKETS_TABLE_SUFFIX)); + || tableName.endsWith(BUCKETS_TABLE_SUFFIX) + || tableName.endsWith(PARTITIONS_TABLE_SUFFIX)); } private void rejectVirtualTableMutation(ObjectPath objectPath, String operation) { @@ -1350,4 +1358,54 @@ private CatalogBaseTable getVirtualBucketsTable(ObjectPath objectPath) t); } } + + /** Creates a read-only virtual table exposing the current partition metadata snapshot. */ + private CatalogBaseTable getVirtualPartitionsTable(ObjectPath objectPath) + throws TableNotExistException, CatalogException { + String virtualTableName = objectPath.getObjectName(); + String baseTableName = + virtualTableName.substring( + 0, virtualTableName.length() - PARTITIONS_TABLE_SUFFIX.length()); + ObjectPath baseObjectPath = new ObjectPath(objectPath.getDatabaseName(), baseTableName); + + try { + // Only validate immutable table metadata here. Partition rows are loaded by the source + // reader at execution time so cached SQL plans observe the latest partitions. + TableInfo tableInfo = admin.getTableInfo(toTablePath(baseObjectPath)).get(); + if (!tableInfo.isPartitioned()) { + throw new UnsupportedOperationException( + String.format( + "The $partitions virtual table is only supported for partitioned " + + "tables, but %s is not partitioned.", + baseObjectPath)); + } + + Schema partitionsSchema = + Schema.newBuilder() + .column("partition_id", BIGINT().notNull()) + .column("partition_name", STRING().notNull()) + .column("bucket_count", INT().notNull()) + .build(); + Map options = new HashMap<>(); + options.put(BOOTSTRAP_SERVERS.key(), bootstrapServers); + options.putAll(securityConfigs); + return CatalogTableAdapter.toCatalogTable( + partitionsSchema, + "Current partition IDs, names, and bucket counts", + Collections.emptyList(), + options); + } catch (UnsupportedOperationException e) { + throw e; + } catch (Exception e) { + Throwable t = ExceptionUtils.stripExecutionException(e); + if (isTableNotExist(t)) { + throw new TableNotExistException(getName(), baseObjectPath); + } + throw new CatalogException( + String.format( + "Failed to get virtual partitions table %s in %s", + objectPath, getName()), + t); + } + } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java index b87639f1052..dbee1b10cf6 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java @@ -29,6 +29,7 @@ import org.apache.fluss.flink.source.BucketsFlinkTableSource; import org.apache.fluss.flink.source.ChangelogFlinkTableSource; import org.apache.fluss.flink.source.FlinkTableSource; +import org.apache.fluss.flink.source.PartitionsFlinkTableSource; import org.apache.fluss.flink.source.reader.LeaseContext; import org.apache.fluss.flink.utils.FlinkConnectorOptionsUtils; import org.apache.fluss.metadata.MergeEngineType; @@ -119,6 +120,10 @@ public DynamicTableSource createDynamicTableSource(Context context) { return createBucketsTableSource(context, tableIdentifier, tableName); } + if (tableName.endsWith(FlinkCatalog.PARTITIONS_TABLE_SUFFIX)) { + return createPartitionsTableSource(context, tableIdentifier, tableName); + } + FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context); final ReadableConfig tableOptions = helper.getOptions(); validateSourceOptions(tableOptions); @@ -492,4 +497,18 @@ private DynamicTableSource createBucketsTableSource( context.getCatalogTable().getOptions(), context.getConfiguration()), tableOutputType); } + + /** Creates a bounded source for a {@code $partitions} virtual table. */ + private DynamicTableSource createPartitionsTableSource( + Context context, ObjectIdentifier tableIdentifier, String tableName) { + String baseTableName = + tableName.substring( + 0, tableName.length() - FlinkCatalog.PARTITIONS_TABLE_SUFFIX.length()); + RowType tableOutputType = (RowType) context.getPhysicalRowDataType().getLogicalType(); + return new PartitionsFlinkTableSource( + TablePath.of(tableIdentifier.getDatabaseName(), baseTableName), + toFlussClientConfig( + context.getCatalogTable().getOptions(), context.getConfiguration()), + tableOutputType); + } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ListPartitionInfosProcedure.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ListPartitionInfosProcedure.java index a069958a027..4415e8b9c41 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ListPartitionInfosProcedure.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ListPartitionInfosProcedure.java @@ -43,7 +43,11 @@ * -- List the partition id, partition name and bucket count of each partition * CALL sys.list_partition_infos('my_db', 'my_table'); * + * + * @deprecated Use the read-only {@code my_table$partitions} virtual table instead. This procedure + * remains available for backward compatibility. */ +@Deprecated public class ListPartitionInfosProcedure extends ProcedureBase { @ProcedureHint( diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BucketsSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BucketsSource.java index f26d0a345fb..b95473d6db6 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BucketsSource.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BucketsSource.java @@ -27,92 +27,28 @@ import org.apache.fluss.utils.ExceptionUtils; import org.apache.flink.api.common.typeinfo.TypeInformation; -import org.apache.flink.api.connector.source.Boundedness; -import org.apache.flink.api.connector.source.ReaderOutput; -import org.apache.flink.api.connector.source.Source; -import org.apache.flink.api.connector.source.SourceReader; -import org.apache.flink.api.connector.source.SourceReaderContext; -import org.apache.flink.api.connector.source.SourceSplit; -import org.apache.flink.api.connector.source.SplitEnumerator; -import org.apache.flink.api.connector.source.SplitEnumeratorContext; -import org.apache.flink.api.java.typeutils.ResultTypeQueryable; -import org.apache.flink.core.io.InputStatus; -import org.apache.flink.core.io.SimpleVersionedSerializer; import org.apache.flink.table.data.GenericArrayData; import org.apache.flink.table.data.GenericRowData; import org.apache.flink.table.data.RowData; import org.apache.flink.table.data.StringData; -import javax.annotation.Nullable; - -import java.io.IOException; -import java.io.Serializable; -import java.nio.ByteBuffer; -import java.util.Collections; import java.util.List; import java.util.OptionalInt; import java.util.OptionalLong; -import java.util.concurrent.CompletableFuture; import java.util.stream.Collectors; -import static org.apache.fluss.utils.Preconditions.checkState; - /** A bounded source that loads bucket metadata once for each execution. */ -final class BucketsSource - implements Source< - RowData, BucketsSource.BucketsSplit, BucketsSource.BucketsEnumeratorState>, - ResultTypeQueryable { +final class BucketsSource extends SingleSplitMetadataSource { private static final long serialVersionUID = 1L; - private final BucketMetadataLoader metadataLoader; - private final TypeInformation producedType; - BucketsSource( TablePath tablePath, Configuration flussConfig, TypeInformation producedType) { this(new AdminBucketMetadataLoader(tablePath, flussConfig), producedType); } - BucketsSource(BucketMetadataLoader metadataLoader, TypeInformation producedType) { - this.metadataLoader = metadataLoader; - this.producedType = producedType; - } - - @Override - public Boundedness getBoundedness() { - return Boundedness.BOUNDED; - } - - @Override - public SplitEnumerator createEnumerator( - SplitEnumeratorContext enumContext) { - return new BucketsSplitEnumerator(enumContext, new BucketsSplit(0)); - } - - @Override - public SplitEnumerator restoreEnumerator( - SplitEnumeratorContext enumContext, BucketsEnumeratorState checkpoint) { - return new BucketsSplitEnumerator(enumContext, checkpoint.pendingSplit); - } - - @Override - public SimpleVersionedSerializer getSplitSerializer() { - return BucketsSplitSerializer.INSTANCE; - } - - @Override - public SimpleVersionedSerializer getEnumeratorCheckpointSerializer() { - return BucketsEnumeratorStateSerializer.INSTANCE; - } - - @Override - public SourceReader createReader(SourceReaderContext readerContext) { - return new BucketsSourceReader(metadataLoader); - } - - @Override - public TypeInformation getProducedType() { - return producedType; + BucketsSource(MetadataLoader metadataLoader, TypeInformation producedType) { + super(metadataLoader, producedType); } static RowData toRowData(BucketInfo bucketInfo) { @@ -139,11 +75,7 @@ private static Integer optionalInt(OptionalInt value) { return value.isPresent() ? value.getAsInt() : null; } - interface BucketMetadataLoader extends Serializable { - List load(); - } - - private static final class AdminBucketMetadataLoader implements BucketMetadataLoader { + private static final class AdminBucketMetadataLoader implements MetadataLoader { private static final long serialVersionUID = 1L; @@ -169,222 +101,4 @@ public List load() { } } } - - static final class BucketsSplit implements SourceSplit, Serializable { - - private static final long serialVersionUID = 1L; - private static final String SPLIT_ID = "buckets"; - - private int nextRow; - - BucketsSplit(int nextRow) { - this.nextRow = nextRow; - } - - @Override - public String splitId() { - return SPLIT_ID; - } - - private void advance() { - nextRow++; - } - } - - static final class BucketsEnumeratorState implements Serializable { - - private static final long serialVersionUID = 1L; - - @Nullable private final BucketsSplit pendingSplit; - - private BucketsEnumeratorState(@Nullable BucketsSplit pendingSplit) { - this.pendingSplit = pendingSplit; - } - } - - private static final class BucketsSplitEnumerator - implements SplitEnumerator { - - private final SplitEnumeratorContext enumContext; - @Nullable private BucketsSplit pendingSplit; - - private BucketsSplitEnumerator( - SplitEnumeratorContext enumContext, - @Nullable BucketsSplit pendingSplit) { - this.enumContext = enumContext; - this.pendingSplit = pendingSplit; - } - - @Override - public void start() {} - - @Override - public void handleSplitRequest(int subtaskId, @Nullable String requesterHostname) { - assignSplitOrSignalFinished(subtaskId); - } - - @Override - public void addSplitsBack(List splits, int subtaskId) { - checkState(pendingSplit == null, "A bucket metadata split is already pending."); - checkState(splits.size() == 1, "Expected one bucket metadata split to be returned."); - pendingSplit = splits.get(0); - } - - @Override - public void addReader(int subtaskId) { - assignSplitOrSignalFinished(subtaskId); - } - - @Override - public BucketsEnumeratorState snapshotState(long checkpointId) { - return new BucketsEnumeratorState(pendingSplit); - } - - @Override - public void close() {} - - private void assignSplitOrSignalFinished(int subtaskId) { - if (pendingSplit != null) { - enumContext.assignSplit(pendingSplit, subtaskId); - pendingSplit = null; - } - enumContext.signalNoMoreSplits(subtaskId); - } - } - - private static final class BucketsSourceReader implements SourceReader { - - private static final CompletableFuture AVAILABLE = - CompletableFuture.completedFuture(null); - - private final BucketMetadataLoader metadataLoader; - - @Nullable private BucketsSplit currentSplit; - @Nullable private List bucketRows; - private CompletableFuture availability = new CompletableFuture<>(); - private boolean noMoreSplits; - - private BucketsSourceReader(BucketMetadataLoader metadataLoader) { - this.metadataLoader = metadataLoader; - } - - @Override - public void start() {} - - @Override - public InputStatus pollNext(ReaderOutput output) { - if (currentSplit == null) { - return noMoreSplits ? InputStatus.END_OF_INPUT : InputStatus.NOTHING_AVAILABLE; - } - - if (bucketRows == null) { - bucketRows = metadataLoader.load(); - } - - if (currentSplit.nextRow >= bucketRows.size()) { - currentSplit = null; - bucketRows = null; - if (!noMoreSplits) { - availability = new CompletableFuture<>(); - } - return noMoreSplits ? InputStatus.END_OF_INPUT : InputStatus.NOTHING_AVAILABLE; - } - - output.collect(bucketRows.get(currentSplit.nextRow)); - currentSplit.advance(); - return InputStatus.MORE_AVAILABLE; - } - - @Override - public List snapshotState(long checkpointId) { - return currentSplit == null - ? Collections.emptyList() - : Collections.singletonList(new BucketsSplit(currentSplit.nextRow)); - } - - @Override - public CompletableFuture isAvailable() { - return currentSplit != null || noMoreSplits ? AVAILABLE : availability; - } - - @Override - public void addSplits(List splits) { - checkState(currentSplit == null, "A bucket metadata split is already being read."); - checkState(splits.size() == 1, "Expected exactly one bucket metadata split."); - currentSplit = splits.get(0); - availability.complete(null); - } - - @Override - public void notifyNoMoreSplits() { - noMoreSplits = true; - availability.complete(null); - } - - @Override - public void close() {} - } - - private static final class BucketsSplitSerializer - implements SimpleVersionedSerializer { - - private static final BucketsSplitSerializer INSTANCE = new BucketsSplitSerializer(); - private static final int VERSION = 1; - - @Override - public int getVersion() { - return VERSION; - } - - @Override - public byte[] serialize(BucketsSplit split) { - return ByteBuffer.allocate(Integer.BYTES).putInt(split.nextRow).array(); - } - - @Override - public BucketsSplit deserialize(int version, byte[] serialized) throws IOException { - checkVersion(version); - return new BucketsSplit(ByteBuffer.wrap(serialized).getInt()); - } - } - - private static final class BucketsEnumeratorStateSerializer - implements SimpleVersionedSerializer { - - private static final BucketsEnumeratorStateSerializer INSTANCE = - new BucketsEnumeratorStateSerializer(); - private static final int VERSION = 1; - - @Override - public int getVersion() { - return VERSION; - } - - @Override - public byte[] serialize(BucketsEnumeratorState state) { - if (state.pendingSplit == null) { - return new byte[] {0}; - } - return ByteBuffer.allocate(1 + Integer.BYTES) - .put((byte) 1) - .putInt(state.pendingSplit.nextRow) - .array(); - } - - @Override - public BucketsEnumeratorState deserialize(int version, byte[] serialized) - throws IOException { - checkVersion(version); - ByteBuffer buffer = ByteBuffer.wrap(serialized); - return buffer.get() == 0 - ? new BucketsEnumeratorState(null) - : new BucketsEnumeratorState(new BucketsSplit(buffer.getInt())); - } - } - - private static void checkVersion(int version) throws IOException { - if (version != 1) { - throw new IOException("Unsupported bucket source serializer version: " + version); - } - } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/PartitionsFlinkTableSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/PartitionsFlinkTableSource.java new file mode 100644 index 00000000000..ea54988a42b --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/PartitionsFlinkTableSource.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.fluss.flink.source; + +import org.apache.fluss.config.Configuration; +import org.apache.fluss.metadata.TablePath; + +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.connector.source.DynamicTableSource; +import org.apache.flink.table.connector.source.ScanTableSource; +import org.apache.flink.table.connector.source.SourceProvider; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.logical.RowType; + +/** A bounded Flink table source for the read-only {@code $partitions} virtual table. */ +public class PartitionsFlinkTableSource implements ScanTableSource { + + private final TablePath tablePath; + private final Configuration flussConfig; + private final RowType producedDataType; + + /** Creates a source for the partition metadata of the given base table. */ + public PartitionsFlinkTableSource( + TablePath tablePath, Configuration flussConfig, RowType producedDataType) { + this.tablePath = tablePath; + this.flussConfig = flussConfig; + this.producedDataType = producedDataType; + } + + @Override + public ChangelogMode getChangelogMode() { + return ChangelogMode.insertOnly(); + } + + @Override + public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { + TypeInformation resultTypeInfo = + scanContext.createTypeInformation(producedDataType); + return SourceProvider.of(new PartitionsSource(tablePath, flussConfig, resultTypeInfo)); + } + + @Override + public DynamicTableSource copy() { + return new PartitionsFlinkTableSource(tablePath, flussConfig, producedDataType); + } + + @Override + public String asSummaryString() { + return "Fluss Partitions Table Source"; + } +} diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/PartitionsSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/PartitionsSource.java new file mode 100644 index 00000000000..7c6dc09ec9d --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/PartitionsSource.java @@ -0,0 +1,84 @@ +/* + * 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.fluss.flink.source; + +import org.apache.fluss.client.Connection; +import org.apache.fluss.client.ConnectionFactory; +import org.apache.fluss.client.admin.Admin; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.exception.FlussRuntimeException; +import org.apache.fluss.metadata.PartitionInfo; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.utils.ExceptionUtils; + +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.StringData; + +import java.util.List; +import java.util.stream.Collectors; + +/** A bounded source that loads partition metadata once for each execution. */ +final class PartitionsSource extends SingleSplitMetadataSource { + + private static final long serialVersionUID = 1L; + + PartitionsSource( + TablePath tablePath, Configuration flussConfig, TypeInformation producedType) { + this(new AdminPartitionMetadataLoader(tablePath, flussConfig), producedType); + } + + PartitionsSource(MetadataLoader metadataLoader, TypeInformation producedType) { + super(metadataLoader, producedType); + } + + static RowData toRowData(PartitionInfo partitionInfo) { + return GenericRowData.of( + partitionInfo.getPartitionId(), + StringData.fromString(partitionInfo.getPartitionName()), + partitionInfo.getBucketCount()); + } + + private static final class AdminPartitionMetadataLoader implements MetadataLoader { + + private static final long serialVersionUID = 1L; + + private final TablePath tablePath; + private final Configuration flussConfig; + + private AdminPartitionMetadataLoader(TablePath tablePath, Configuration flussConfig) { + this.tablePath = tablePath; + this.flussConfig = flussConfig; + } + + @Override + public List load() { + try (Connection connection = ConnectionFactory.createConnection(flussConfig); + Admin admin = connection.getAdmin()) { + return admin.listPartitionInfos(tablePath).get().stream() + .map(PartitionsSource::toRowData) + .collect(Collectors.toList()); + } catch (Exception e) { + throw new FlussRuntimeException( + String.format("Failed to list partitions for table %s.", tablePath), + ExceptionUtils.stripExecutionException(e)); + } + } + } +} diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/SingleSplitMetadataSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/SingleSplitMetadataSource.java new file mode 100644 index 00000000000..3e71a8cc104 --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/SingleSplitMetadataSource.java @@ -0,0 +1,323 @@ +/* + * 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.fluss.flink.source; + +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.connector.source.Boundedness; +import org.apache.flink.api.connector.source.ReaderOutput; +import org.apache.flink.api.connector.source.Source; +import org.apache.flink.api.connector.source.SourceReader; +import org.apache.flink.api.connector.source.SourceReaderContext; +import org.apache.flink.api.connector.source.SourceSplit; +import org.apache.flink.api.connector.source.SplitEnumerator; +import org.apache.flink.api.connector.source.SplitEnumeratorContext; +import org.apache.flink.api.java.typeutils.ResultTypeQueryable; +import org.apache.flink.core.io.InputStatus; +import org.apache.flink.core.io.SimpleVersionedSerializer; +import org.apache.flink.table.data.RowData; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.io.Serializable; +import java.nio.ByteBuffer; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CompletableFuture; + +import static org.apache.fluss.utils.Preconditions.checkState; + +/** A bounded singleton-split source that loads metadata once for each execution. */ +class SingleSplitMetadataSource + implements Source< + RowData, + SingleSplitMetadataSource.MetadataSplit, + SingleSplitMetadataSource.MetadataEnumeratorState>, + ResultTypeQueryable { + + private static final long serialVersionUID = 1L; + + private final MetadataLoader metadataLoader; + private final TypeInformation producedType; + + SingleSplitMetadataSource( + MetadataLoader metadataLoader, TypeInformation producedType) { + this.metadataLoader = metadataLoader; + this.producedType = producedType; + } + + @Override + public Boundedness getBoundedness() { + return Boundedness.BOUNDED; + } + + @Override + public SplitEnumerator createEnumerator( + SplitEnumeratorContext enumContext) { + return new MetadataSplitEnumerator(enumContext, new MetadataSplit(0)); + } + + @Override + public SplitEnumerator restoreEnumerator( + SplitEnumeratorContext enumContext, MetadataEnumeratorState checkpoint) { + return new MetadataSplitEnumerator(enumContext, checkpoint.pendingSplit); + } + + @Override + public SimpleVersionedSerializer getSplitSerializer() { + return MetadataSplitSerializer.INSTANCE; + } + + @Override + public SimpleVersionedSerializer getEnumeratorCheckpointSerializer() { + return MetadataEnumeratorStateSerializer.INSTANCE; + } + + @Override + public SourceReader createReader(SourceReaderContext readerContext) { + return new MetadataSourceReader(metadataLoader); + } + + @Override + public TypeInformation getProducedType() { + return producedType; + } + + interface MetadataLoader extends Serializable { + List load(); + } + + static final class MetadataSplit implements SourceSplit, Serializable { + + private static final long serialVersionUID = 1L; + private static final String SPLIT_ID = "metadata"; + + private int nextRow; + + MetadataSplit(int nextRow) { + this.nextRow = nextRow; + } + + @Override + public String splitId() { + return SPLIT_ID; + } + + private void advance() { + nextRow++; + } + } + + static final class MetadataEnumeratorState implements Serializable { + + private static final long serialVersionUID = 1L; + + @Nullable private final MetadataSplit pendingSplit; + + private MetadataEnumeratorState(@Nullable MetadataSplit pendingSplit) { + this.pendingSplit = pendingSplit; + } + } + + private static final class MetadataSplitEnumerator + implements SplitEnumerator { + + private final SplitEnumeratorContext enumContext; + @Nullable private MetadataSplit pendingSplit; + + private MetadataSplitEnumerator( + SplitEnumeratorContext enumContext, + @Nullable MetadataSplit pendingSplit) { + this.enumContext = enumContext; + this.pendingSplit = pendingSplit; + } + + @Override + public void start() {} + + @Override + public void handleSplitRequest(int subtaskId, @Nullable String requesterHostname) { + assignSplitOrSignalFinished(subtaskId); + } + + @Override + public void addSplitsBack(List splits, int subtaskId) { + checkState(pendingSplit == null, "A metadata split is already pending."); + checkState(splits.size() == 1, "Expected one metadata split to be returned."); + pendingSplit = splits.get(0); + } + + @Override + public void addReader(int subtaskId) { + assignSplitOrSignalFinished(subtaskId); + } + + @Override + public MetadataEnumeratorState snapshotState(long checkpointId) { + return new MetadataEnumeratorState(pendingSplit); + } + + @Override + public void close() {} + + private void assignSplitOrSignalFinished(int subtaskId) { + if (pendingSplit != null) { + enumContext.assignSplit(pendingSplit, subtaskId); + pendingSplit = null; + } + enumContext.signalNoMoreSplits(subtaskId); + } + } + + private static final class MetadataSourceReader + implements SourceReader { + + private static final CompletableFuture AVAILABLE = + CompletableFuture.completedFuture(null); + + private final MetadataLoader metadataLoader; + + @Nullable private MetadataSplit currentSplit; + @Nullable private List metadataRows; + private CompletableFuture availability = new CompletableFuture<>(); + private boolean noMoreSplits; + + private MetadataSourceReader(MetadataLoader metadataLoader) { + this.metadataLoader = metadataLoader; + } + + @Override + public void start() {} + + @Override + public InputStatus pollNext(ReaderOutput output) { + if (currentSplit == null) { + return noMoreSplits ? InputStatus.END_OF_INPUT : InputStatus.NOTHING_AVAILABLE; + } + + if (metadataRows == null) { + metadataRows = metadataLoader.load(); + } + + if (currentSplit.nextRow >= metadataRows.size()) { + currentSplit = null; + metadataRows = null; + if (!noMoreSplits) { + availability = new CompletableFuture<>(); + } + return noMoreSplits ? InputStatus.END_OF_INPUT : InputStatus.NOTHING_AVAILABLE; + } + + output.collect(metadataRows.get(currentSplit.nextRow)); + currentSplit.advance(); + return InputStatus.MORE_AVAILABLE; + } + + @Override + public List snapshotState(long checkpointId) { + return currentSplit == null + ? Collections.emptyList() + : Collections.singletonList(new MetadataSplit(currentSplit.nextRow)); + } + + @Override + public CompletableFuture isAvailable() { + return currentSplit != null || noMoreSplits ? AVAILABLE : availability; + } + + @Override + public void addSplits(List splits) { + checkState(currentSplit == null, "A metadata split is already being read."); + checkState(splits.size() == 1, "Expected exactly one metadata split."); + currentSplit = splits.get(0); + availability.complete(null); + } + + @Override + public void notifyNoMoreSplits() { + noMoreSplits = true; + availability.complete(null); + } + + @Override + public void close() {} + } + + private static final class MetadataSplitSerializer + implements SimpleVersionedSerializer { + + private static final MetadataSplitSerializer INSTANCE = new MetadataSplitSerializer(); + private static final int VERSION = 1; + + @Override + public int getVersion() { + return VERSION; + } + + @Override + public byte[] serialize(MetadataSplit split) { + return ByteBuffer.allocate(Integer.BYTES).putInt(split.nextRow).array(); + } + + @Override + public MetadataSplit deserialize(int version, byte[] serialized) throws IOException { + checkVersion(version); + return new MetadataSplit(ByteBuffer.wrap(serialized).getInt()); + } + } + + private static final class MetadataEnumeratorStateSerializer + implements SimpleVersionedSerializer { + + private static final MetadataEnumeratorStateSerializer INSTANCE = + new MetadataEnumeratorStateSerializer(); + private static final int VERSION = 1; + + @Override + public int getVersion() { + return VERSION; + } + + @Override + public byte[] serialize(MetadataEnumeratorState state) { + if (state.pendingSplit == null) { + return new byte[] {0}; + } + return ByteBuffer.allocate(1 + Integer.BYTES) + .put((byte) 1) + .putInt(state.pendingSplit.nextRow) + .array(); + } + + @Override + public MetadataEnumeratorState deserialize(int version, byte[] serialized) + throws IOException { + checkVersion(version); + ByteBuffer buffer = ByteBuffer.wrap(serialized); + return buffer.get() == 0 + ? new MetadataEnumeratorState(null) + : new MetadataEnumeratorState(new MetadataSplit(buffer.getInt())); + } + } + + private static void checkVersion(int version) throws IOException { + if (version != 1) { + throw new IOException("Unsupported metadata source serializer version: " + version); + } + } +} diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java index 66202310a58..88ca55bd9c6 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java @@ -82,6 +82,7 @@ import static org.apache.fluss.testutils.common.CommonTestUtils.retry; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.groups.Tuple.tuple; /** IT case for {@link org.apache.fluss.flink.catalog.FlinkCatalog}. */ abstract class FlinkCatalogITCase { @@ -1451,6 +1452,88 @@ void testBucketsVirtualTable() throws Exception { .hasMessageContaining("read-only virtual table"); } + @Test + @MultiVersionTest + void testPartitionsVirtualTable() throws Exception { + String tableName = "table_for_partitions"; + ObjectPath partitionsPath = + new ObjectPath(DEFAULT_DB, tableName + FlinkCatalog.PARTITIONS_TABLE_SUFFIX); + tEnv.executeSql( + "CREATE TABLE " + + tableName + + " (id INT, region STRING) PARTITIONED BY (region) " + + "WITH ('bucket.num' = '2')"); + tEnv.executeSql("ALTER TABLE " + tableName + " ADD PARTITION (region = 'old')"); + tEnv.executeSql("ALTER TABLE " + tableName + " SET ('bucket.num' = '4')"); + tEnv.executeSql("ALTER TABLE " + tableName + " ADD PARTITION (region = 'new')"); + + CatalogTable partitionsTable = (CatalogTable) catalog.getTable(partitionsPath); + Schema expectedSchema = + Schema.newBuilder() + .column("partition_id", DataTypes.BIGINT().notNull()) + .column("partition_name", DataTypes.STRING().notNull()) + .column("bucket_count", DataTypes.INT().notNull()) + .build(); + assertThat(partitionsTable.getUnresolvedSchema()).isEqualTo(expectedSchema); + assertThat(partitionsTable.getPartitionKeys()).isEmpty(); + assertThat(catalog.tableExists(partitionsPath)).isTrue(); + assertThat(collectRows("DESCRIBE " + tableName + FlinkCatalog.PARTITIONS_TABLE_SUFFIX)) + .extracting(row -> row.getField(0)) + .containsExactly("partition_id", "partition_name", "bucket_count"); + + retry( + Duration.ofMinutes(1), + () -> { + List rows = + collectRows( + "SELECT partition_id, partition_name, bucket_count FROM " + + tableName + + FlinkCatalog.PARTITIONS_TABLE_SUFFIX); + assertThat(rows).hasSize(2); + assertThat(rows).allSatisfy(row -> assertThat(row.getField(0)).isNotNull()); + assertThat(rows) + .extracting(row -> row.getField(1), row -> row.getField(2)) + .containsExactlyInAnyOrder(tuple("old", 2), tuple("new", 4)); + + assertThat( + collectRows( + "SELECT partition_name FROM " + + tableName + + FlinkCatalog.PARTITIONS_TABLE_SUFFIX + + " WHERE bucket_count = 4")) + .extracting(row -> row.getField(0)) + .containsExactly("new"); + + assertThat( + collectRows( + "SELECT bucket_count, COUNT(*) FROM " + + tableName + + FlinkCatalog.PARTITIONS_TABLE_SUFFIX + + " GROUP BY bucket_count")) + .extracting(row -> row.getField(0), row -> row.getField(1)) + .containsExactlyInAnyOrder(tuple(2, 1L), tuple(4, 1L)); + }); + + assertThatThrownBy( + () -> + catalog.getTable( + new ObjectPath(DEFAULT_DB, "missing_table$partitions"))) + .isInstanceOf(TableNotExistException.class); + assertThatThrownBy(() -> catalog.dropTable(partitionsPath, false)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("read-only virtual table"); + + tEnv.executeSql("CREATE TABLE non_partitioned_for_partitions (id INT)"); + assertThatThrownBy( + () -> + catalog.getTable( + new ObjectPath( + DEFAULT_DB, + "non_partitioned_for_partitions$partitions"))) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("only supported for partitioned tables"); + } + private List collectRows(String sql) throws Exception { try (CloseableIterator rows = tEnv.executeSql(sql).collect()) { return CollectionUtil.iteratorToList(rows); diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkTableFactoryTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkTableFactoryTest.java index 56f74fcebe9..7d706bd4748 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkTableFactoryTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkTableFactoryTest.java @@ -26,6 +26,7 @@ import org.apache.fluss.flink.source.BucketsFlinkTableSource; import org.apache.fluss.flink.source.ChangelogFlinkTableSource; import org.apache.fluss.flink.source.FlinkTableSource; +import org.apache.fluss.flink.source.PartitionsFlinkTableSource; import org.apache.fluss.flink.source.lookup.FlinkAsyncLookupFunction; import org.apache.fluss.flink.source.lookup.FlinkLookupFunction; import org.apache.fluss.testutils.common.MultiVersionTest; @@ -89,14 +90,21 @@ abstract class FlinkTableFactoryTest { public static final ObjectIdentifier BUCKETS_TABLE_IDENTIFIER = ObjectIdentifier.of("default", "default", "t1" + FlinkCatalog.BUCKETS_TABLE_SUFFIX); + public static final ObjectIdentifier PARTITIONS_TABLE_IDENTIFIER = + ObjectIdentifier.of("default", "default", "t1" + FlinkCatalog.PARTITIONS_TABLE_SUFFIX); + @Test void testResolveToLakeObjectName() { assertThat(FlinkCatalog.isVirtualTableName("fluss_table$buckets")).isTrue(); assertThat(FlinkCatalog.isVirtualTableName("fluss_table$lake$buckets")).isFalse(); + assertThat(FlinkCatalog.isVirtualTableName("fluss_table$partitions")).isTrue(); + assertThat(FlinkCatalog.isVirtualTableName("fluss_table$lake$partitions")).isFalse(); assertThat(FlinkCatalog.resolveToLakeObjectName(null, "fluss_table$lake")) .isEqualTo("fluss_table"); assertThat(FlinkCatalog.resolveToLakeObjectName(null, "fluss_table$lake$snapshots")) .isEqualTo("fluss_table$snapshots"); + assertThat(FlinkCatalog.resolveToLakeObjectName(null, "fluss_table$lake$partitions")) + .isEqualTo("fluss_table$partitions"); assertThat(FlinkCatalog.resolveToLakeObjectName("custom_lake_table", "fluss_table$lake")) .isEqualTo("custom_lake_table"); assertThat( @@ -297,6 +305,15 @@ void testVirtualLogTableSourceSupportsBatchMode() { Collections.emptyMap(), configuration)) .isInstanceOf(BucketsFlinkTableSource.class); + + assertThat( + createTableSource( + PARTITIONS_TABLE_IDENTIFIER, + createPartitionsSchema(), + properties, + Collections.emptyMap(), + configuration)) + .isInstanceOf(PartitionsFlinkTableSource.class); } @Test @@ -329,6 +346,15 @@ void testVirtualTableSources() { Collections.emptyMap(), new Configuration()); assertThat(bucketsSource).isInstanceOf(BucketsFlinkTableSource.class); + + DynamicTableSource partitionsSource = + createTableSource( + PARTITIONS_TABLE_IDENTIFIER, + createPartitionsSchema(), + properties, + Collections.emptyMap(), + new Configuration()); + assertThat(partitionsSource).isInstanceOf(PartitionsFlinkTableSource.class); } @Test @@ -341,6 +367,15 @@ void testVirtualTableSinkIsRejected() { getBasicOptions())) .isInstanceOf(UnsupportedOperationException.class) .hasMessageContaining("read-only virtual table"); + + assertThatThrownBy( + () -> + createTableSink( + PARTITIONS_TABLE_IDENTIFIER, + createPartitionsSchema(), + getBasicOptions())) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("read-only virtual table"); } @Test @@ -476,6 +511,16 @@ private ResolvedSchema createBucketsSchema() { null); } + private ResolvedSchema createPartitionsSchema() { + return new ResolvedSchema( + Arrays.asList( + Column.physical("partition_id", DataTypes.BIGINT().notNull()), + Column.physical("partition_name", DataTypes.STRING().notNull()), + Column.physical("bucket_count", DataTypes.INT().notNull())), + Collections.emptyList(), + null); + } + private static Map getBasicOptions() { Map options = new HashMap<>(); options.put("connector", "fluss"); diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BucketsFlinkTableSourceTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BucketsFlinkTableSourceTest.java index 008beee5022..46950066fee 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BucketsFlinkTableSourceTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BucketsFlinkTableSourceTest.java @@ -98,10 +98,12 @@ void testRestoredReaderResumesFromSplitOffset() throws Exception { return Arrays.asList(GenericRowData.of(1), GenericRowData.of(2)); }, TypeInformation.of(RowData.class)); - List checkpoint; + List checkpoint; - try (SourceReader reader = source.createReader(null)) { - reader.addSplits(Collections.singletonList(new BucketsSource.BucketsSplit(0))); + try (SourceReader reader = + source.createReader(null)) { + reader.addSplits( + Collections.singletonList(new SingleSplitMetadataSource.MetadataSplit(0))); reader.notifyNoMoreSplits(); TestingReaderOutput output = new TestingReaderOutput<>(); @@ -110,7 +112,7 @@ void testRestoredReaderResumesFromSplitOffset() throws Exception { checkpoint = reader.snapshotState(1L); } - try (SourceReader restoredReader = + try (SourceReader restoredReader = source.createReader(null)) { restoredReader.addSplits(checkpoint); restoredReader.notifyNoMoreSplits(); @@ -125,9 +127,11 @@ void testRestoredReaderResumesFromSplitOffset() throws Exception { } private int readSingleValue(BucketsSource source) throws Exception { - try (SourceReader reader = source.createReader(null)) { + try (SourceReader reader = + source.createReader(null)) { assertThat(source.getBoundedness()).isEqualTo(Boundedness.BOUNDED); - reader.addSplits(Collections.singletonList(new BucketsSource.BucketsSplit(0))); + reader.addSplits( + Collections.singletonList(new SingleSplitMetadataSource.MetadataSplit(0))); reader.notifyNoMoreSplits(); TestingReaderOutput output = new TestingReaderOutput<>(); diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/PartitionsFlinkTableSourceTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/PartitionsFlinkTableSourceTest.java new file mode 100644 index 00000000000..4a59e5c519f --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/PartitionsFlinkTableSourceTest.java @@ -0,0 +1,89 @@ +/* + * 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.fluss.flink.source; + +import org.apache.fluss.metadata.PartitionInfo; +import org.apache.fluss.metadata.ResolvedPartitionSpec; + +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.connector.source.SourceReader; +import org.apache.flink.connector.testutils.source.reader.TestingReaderOutput; +import org.apache.flink.core.io.InputStatus; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Test for {@link PartitionsFlinkTableSource}. */ +class PartitionsFlinkTableSourceTest { + + @Test + void testConvertPartitionInfoToInternalRow() { + PartitionInfo partitionInfo = + new PartitionInfo( + 10L, + new ResolvedPartitionSpec( + Collections.singletonList("region"), + Collections.singletonList("cn")), + null, + 4); + + RowData row = PartitionsSource.toRowData(partitionInfo); + + assertThat(row.getArity()).isEqualTo(3); + assertThat(row.getLong(0)).isEqualTo(10L); + assertThat(row.getString(1).toString()).isEqualTo("cn"); + assertThat(row.getInt(2)).isEqualTo(4); + } + + @Test + void testLoadsFreshMetadataForEveryReaderExecution() throws Exception { + AtomicInteger loadCount = new AtomicInteger(); + PartitionsSource source = + new PartitionsSource( + () -> + Collections.singletonList( + GenericRowData.of(loadCount.incrementAndGet())), + TypeInformation.of(RowData.class)); + + assertThat(loadCount).hasValue(0); + assertThat(readSingleValue(source)).isEqualTo(1); + assertThat(loadCount).hasValue(1); + assertThat(readSingleValue(source)).isEqualTo(2); + assertThat(loadCount).hasValue(2); + } + + private int readSingleValue(PartitionsSource source) throws Exception { + try (SourceReader reader = + source.createReader(null)) { + reader.addSplits( + Collections.singletonList(new SingleSplitMetadataSource.MetadataSplit(0))); + reader.notifyNoMoreSplits(); + TestingReaderOutput output = new TestingReaderOutput<>(); + + assertThat(reader.pollNext(output)).isEqualTo(InputStatus.MORE_AVAILABLE); + assertThat(reader.pollNext(output)).isEqualTo(InputStatus.END_OF_INPUT); + assertThat(output.getEmittedRecords()).hasSize(1); + return output.getEmittedRecords().get(0).getInt(0); + } + } +} diff --git a/website/docs/engine-flink/procedures.md b/website/docs/engine-flink/procedures.md index 870227f0d50..9ed41e3b8b7 100644 --- a/website/docs/engine-flink/procedures.md +++ b/website/docs/engine-flink/procedures.md @@ -696,6 +696,18 @@ Fluss provides procedures to inspect partition-level metadata of partitioned tab List the partition id, partition name and bucket count of every partition of a partitioned table. +:::warning Deprecated +Use the [`$partitions`](../table-design/virtual-tables.md#partitions-table) virtual table instead: + +```sql title="Flink SQL" +SELECT partition_id, partition_name, bucket_count +FROM my_db.orders$partitions; +``` + +This procedure remains available for backward compatibility and will be removed in a future +release. +::: + For a partitioned table, `ALTER TABLE ... SET ('bucket.num' = N)` only changes the bucket count for newly created partitions; existing partitions keep their original bucket counts. The table-level `'bucket.num'` shown by `SHOW CREATE TABLE` is only the default for new partitions, so use this procedure to see the actual bucket count of each partition. **Syntax:** diff --git a/website/docs/table-design/data-distribution/bucketing.md b/website/docs/table-design/data-distribution/bucketing.md index 75b72ec0c04..092d369103b 100644 --- a/website/docs/table-design/data-distribution/bucketing.md +++ b/website/docs/table-design/data-distribution/bucketing.md @@ -111,10 +111,11 @@ The `SELECT` returns all five rows from the mixed-bucket-count table — three f SHOW CREATE TABLE order_events; ``` -To inspect the actual bucket count of each partition, use the [`sys.list_partition_infos`](engine-flink/procedures.md#list_partition_infos) procedure: +To inspect the actual bucket count of each partition, query the [`$partitions`](../virtual-tables.md#partitions-table) virtual table: ```sql title="Flink SQL" -CALL sys.list_partition_infos('my_db', 'order_events'); +SELECT partition_id, partition_name, bucket_count +FROM my_db.order_events$partitions; ``` It returns one row per partition with the partition id, the partition name, and the partition's actual bucket count. For the table above after the rescale, the result looks like: diff --git a/website/docs/table-design/virtual-tables.md b/website/docs/table-design/virtual-tables.md index 203aef23ea2..e2e1a1bcc85 100644 --- a/website/docs/table-design/virtual-tables.md +++ b/website/docs/table-design/virtual-tables.md @@ -15,12 +15,13 @@ Fluss supports the following virtual table types: | [Changelog](#changelog-table) | `$changelog` | Provides access to the raw changelog stream with metadata | Primary Key Tables, Log Tables | | [Binlog](#binlog-table) | `$binlog` | Provides binlog format with before/after metadata | Primary Key Tables only | | [Buckets](#buckets-table) | `$buckets` | Provides replica placement and leader/ISR metadata | Primary Key Tables, Log Tables | +| [Partitions](#partitions-table) | `$partitions` | Provides partition IDs, names, and bucket counts | Partitioned Tables | ## Flink Runtime Modes Both `$changelog` and `$binlog` can be queried in Flink streaming or batch runtime. -The `$buckets` table always returns a bounded metadata snapshot, including when the Flink runtime -mode is streaming. +The `$buckets` and `$partitions` tables always return bounded metadata snapshots, including when +the Flink runtime mode is streaming. | Runtime Mode | Behavior | |--------------|----------| @@ -55,6 +56,41 @@ LIMIT 100; On Flink 1.18, a predicate on `_commit_timestamp` (or on any other `TIMESTAMP_LTZ` column) returns wrong results when the session time zone is not UTC. Flink 1.18 rebuilds such a predicate using the session time zone instead of UTC after the filter is pushed into the source, which shifts the literal by the zone offset and typically makes the query return no rows. This is [FLINK-35318](https://issues.apache.org/jira/browse/FLINK-35318), fixed in Flink 1.19.2 and 1.20.0 and never backported to 1.18. On Flink 1.18, either avoid `TIMESTAMP_LTZ` predicates or set `table.local-time-zone` to `UTC`. ::: +## Partitions Table + +The `$partitions` virtual table provides read-only access to the current metadata of every regular +partition in a partitioned table. Append `$partitions` to the base table name: + +```sql title="Flink SQL" +SELECT * FROM orders$partitions; +``` + +### Schema + +| Column | Type | Description | +|--------|------|-------------| +| `partition_id` | BIGINT NOT NULL | The globally unique partition ID | +| `partition_name` | STRING NOT NULL | The partition name | +| `bucket_count` | INT NOT NULL | The actual bucket count of the partition | + +The table can be inspected and queried with standard SQL: + +```sql title="Flink SQL" +DESCRIBE orders$partitions; + +SELECT partition_name, bucket_count +FROM orders$partitions +WHERE bucket_count > 2; +``` + +After `ALTER TABLE ... SET ('bucket.num' = N)`, existing partitions keep their original bucket +counts and only newly created partitions use the new value. Query `$partitions` to inspect the +effective bucket count of each partition instead of relying on the table-level default shown by +`SHOW CREATE TABLE`. + +The current implementation obtains the complete partition metadata snapshot before Flink evaluates +the predicate. Predicate pushdown is not supported yet. + ## Buckets Table The `$buckets` virtual table provides read-only access to the current replica placement and