You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ja...@apache.org on 2025/11/30 18:36:02 UTC
(fluss) branch ci-add-column created (now ff3db43d4)
This is an automated email from the ASF dual-hosted git repository.
jark pushed a change to branch ci-add-column
in repository https://gitbox.apache.org/repos/asf/fluss.git
at ff3db43d4 WIP3
This branch includes the following new commits:
new e715276f2 WIP
new 6295cf7d9 WIP1
new df0abce84 server projection cache
new 78eedf86c WIP
new 5d65a20b7 WIP2
new aa31e1c4c revert FlinkAsFlussRow changes
new ff3db43d4 WIP3
The 7 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(fluss) 03/07: server projection cache
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch ci-add-column
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit df0abce847b5d0dffbe54c670bd0e4bc4a5b5b6d
Author: Jark Wu <ja...@apache.org>
AuthorDate: Mon Dec 1 00:00:31 2025 +0800
server projection cache
---
.../scanner/log/DefaultCompletedFetchTest.java | 3 +-
.../org/apache/fluss/record/FileLogProjection.java | 105 ++++++++-------------
.../fluss/record/ProjectionPushdownCache.java | 86 +++++++++++++++++
.../apache/fluss/record/FileLogProjectionTest.java | 55 ++++-------
.../org/apache/fluss/server/log/FetchParams.java | 6 +-
.../fluss/server/replica/ReplicaManager.java | 5 +-
.../org/apache/fluss/server/kv/KvTabletTest.java | 11 ++-
.../apache/fluss/server/log/FetchParamsTest.java | 14 ++-
.../apache/fluss/server/replica/ReplicaTest.java | 16 +++-
9 files changed, 182 insertions(+), 119 deletions(-)
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/DefaultCompletedFetchTest.java b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/DefaultCompletedFetchTest.java
index bfd686dc1..d487a2d46 100644
--- a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/DefaultCompletedFetchTest.java
+++ b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/DefaultCompletedFetchTest.java
@@ -29,6 +29,7 @@ import org.apache.fluss.record.FileLogProjection;
import org.apache.fluss.record.FileLogRecords;
import org.apache.fluss.record.LogRecordReadContext;
import org.apache.fluss.record.MemoryLogRecords;
+import org.apache.fluss.record.ProjectionPushdownCache;
import org.apache.fluss.record.TestingSchemaGetter;
import org.apache.fluss.row.InternalRow;
import org.apache.fluss.rpc.entity.FetchLogResultForBucket;
@@ -265,7 +266,7 @@ public class DefaultCompletedFetchTest {
rowType, DEFAULT_SCHEMA_ID, 0L, 1000L, magic, objects, LogFormat.ARROW));
fileLogRecords.flush();
- FileLogProjection fileLogProjection = new FileLogProjection();
+ FileLogProjection fileLogProjection = new FileLogProjection(new ProjectionPushdownCache());
fileLogProjection.setCurrentProjection(
DATA2_TABLE_ID,
testingSchemaGetter,
diff --git a/fluss-common/src/main/java/org/apache/fluss/record/FileLogProjection.java b/fluss-common/src/main/java/org/apache/fluss/record/FileLogProjection.java
index 11c50bb3d..546275f55 100644
--- a/fluss-common/src/main/java/org/apache/fluss/record/FileLogProjection.java
+++ b/fluss-common/src/main/java/org/apache/fluss/record/FileLogProjection.java
@@ -54,7 +54,6 @@ import java.util.BitSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
import static org.apache.fluss.record.DefaultLogRecordBatch.APPEND_ONLY_FLAG_MASK;
import static org.apache.fluss.record.LogRecordBatchFormat.LENGTH_OFFSET;
@@ -71,7 +70,6 @@ import static org.apache.fluss.record.LogRecordBatchFormat.recordsCountOffset;
import static org.apache.fluss.record.LogRecordBatchFormat.schemaIdOffset;
import static org.apache.fluss.utils.FileUtils.readFully;
import static org.apache.fluss.utils.FileUtils.readFullyOrFail;
-import static org.apache.fluss.utils.Preconditions.checkNotNull;
import static org.apache.fluss.utils.Preconditions.checkState;
/** Column projection util on Arrow format {@link FileLogRecords}. */
@@ -85,8 +83,8 @@ public class FileLogProjection {
private static final int ARROW_HEADER_SIZE =
ARROW_IPC_CONTINUATION_LENGTH + ARROW_IPC_METADATA_SIZE_LENGTH;
- final Map<ProjectionKey, ProjectionInfo> projectionsCache = new HashMap<>();
- ProjectionInfo currentProjection;
+ // the projection cache shared in the TabletServer
+ private final ProjectionPushdownCache projectionsCache;
// shared resources for multiple projections
private final ByteArrayOutputStream outputStream;
@@ -103,9 +101,10 @@ public class FileLogProjection {
private SchemaGetter schemaGetter;
private long tableId;
private ArrowCompressionInfo compressionInfo;
- private int[] selectedFieldPositions;
+ private int[] selectedFieldIds;
- public FileLogProjection() {
+ public FileLogProjection(ProjectionPushdownCache projectionsCache) {
+ this.projectionsCache = projectionsCache;
this.outputStream = new ByteArrayOutputStream();
this.writeChannel = new WriteChannel(Channels.newChannel(outputStream));
// fluss use little endian for encoding log records batch
@@ -122,14 +121,7 @@ public class FileLogProjection {
this.tableId = tableId;
this.schemaGetter = schemaGetter;
this.compressionInfo = compressionInfo;
-
- // Currently, only add last column is supported.Thus selectedFieldPositions is always same
- // from same selectedFieldIds.
- // TODO: if support drop column or add column in middle, this selectedFieldPositions should
- // be re-calculated for each schema.
- this.selectedFieldPositions =
- selectedFieldPositions(
- schemaGetter.getLatestSchemaInfo().getSchema(), selectedFieldIds);
+ this.selectedFieldIds = selectedFieldIds;
}
/**
@@ -144,6 +136,8 @@ public class FileLogProjection {
MultiBytesView.Builder builder = MultiBytesView.builder();
int position = start;
+ ProjectionInfo currentProjection = null;
+ short prevSchemaId = -1;
// The condition is an optimization to avoid read log header when there is no enough bytes,
// So we use V0 header size here for a conservative judgment. In the end, the condition
// of (position >= end - recordBatchHeaderSize) will ensure the final correctness.
@@ -162,8 +156,12 @@ public class FileLogProjection {
int recordBatchHeaderSize = recordBatchHeaderSize(magic);
int batchSizeInBytes = LOG_OVERHEAD + logHeaderBuffer.getInt(LENGTH_OFFSET);
short schemaId = logHeaderBuffer.getShort(schemaIdOffset(magic));
- setCurrentSchema(schemaId);
- checkNotNull(currentProjection, "There is no projection registered yet.");
+
+ // reuse projection in the current log file
+ if (currentProjection == null || prevSchemaId != schemaId) {
+ prevSchemaId = schemaId;
+ currentProjection = getOrCreateProjectionInfo(schemaId);
+ }
if (position > end - batchSizeInBytes) {
// the remaining bytes in the file are not enough to read a full batch
@@ -405,20 +403,22 @@ public class FileLogProjection {
return logHeaderBuffer;
}
- private void setCurrentSchema(short schemaId) {
+ private ProjectionInfo getOrCreateProjectionInfo(short schemaId) {
+ ProjectionInfo cachedProjection =
+ projectionsCache.getProjectionInfo(tableId, schemaId, selectedFieldIds);
+ if (cachedProjection == null) {
+ cachedProjection = createProjectionInfo(schemaId, selectedFieldIds);
+ projectionsCache.setProjectionInfo(
+ tableId, schemaId, selectedFieldIds, cachedProjection);
+ }
+ return cachedProjection;
+ }
+
+ private ProjectionInfo createProjectionInfo(short schemaId, int[] selectedFieldIds) {
org.apache.fluss.metadata.Schema schema = schemaGetter.getSchema(schemaId);
+ int[] selectedFieldPositions =
+ selectedFieldPositions(schemaGetter.getSchema(schemaId), selectedFieldIds);
RowType rowType = schema.getRowType();
- ProjectionKey projectionKey = new ProjectionKey(tableId, schemaId);
- if (projectionsCache.containsKey(projectionKey)) {
- // the schema and projection should identical for the same table id.
- currentProjection = projectionsCache.get(projectionKey);
- if (!Arrays.equals(currentProjection.selectedFieldPositions, selectedFieldPositions)
- || !currentProjection.schema.equals(rowType)) {
- throw new InvalidColumnProjectionException(
- "The schema and projection should be identical for the same table id.");
- }
- return;
- }
// initialize the projection util information
Schema arrowSchema = ArrowUtils.toArrowSchema(rowType);
@@ -451,16 +451,14 @@ public class FileLogProjection {
CompressionUtil.createBodyCompression(compressionInfo.createCompressionCodec());
int metadataLength =
ArrowUtils.estimateArrowMetadataLength(projectedArrowSchema, bodyCompression);
- currentProjection =
- new ProjectionInfo(
- nodesProjection,
- buffersProjection,
- bufferIndex,
- rowType,
- metadataLength,
- bodyCompression,
- selectedFieldPositions);
- projectionsCache.put(projectionKey, currentProjection);
+ return new ProjectionInfo(
+ nodesProjection,
+ buffersProjection,
+ bufferIndex,
+ rowType,
+ metadataLength,
+ bodyCompression,
+ selectedFieldPositions);
}
int[] selectedFieldPositions(org.apache.fluss.metadata.Schema schema, int[] projectedFields) {
@@ -478,7 +476,8 @@ public class FileLogProjection {
if (position == null) {
throw new InvalidColumnProjectionException(
String.format(
- "Projected field id %s is not contains in %s", fieldId, columnIds));
+ "Projected field id %s is not contained in %s",
+ fieldId, columnIds));
}
selectedFieldPositions[i] = position;
@@ -493,31 +492,7 @@ public class FileLogProjection {
return selectedFieldPositions;
}
- static final class ProjectionKey {
- private final long tableId;
- private final short schemaId;
-
- ProjectionKey(long tableId, short schemaId) {
- this.tableId = tableId;
- this.schemaId = schemaId;
- }
-
- @Override
- public boolean equals(Object o) {
- if (!(o instanceof ProjectionKey)) {
- return false;
- }
- ProjectionKey that = (ProjectionKey) o;
- return tableId == that.tableId && schemaId == that.schemaId;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(tableId, schemaId);
- }
- }
-
- static final class ProjectionInfo {
+ public static final class ProjectionInfo {
final BitSet nodesProjection;
final BitSet buffersProjection;
final int bufferCount;
@@ -545,7 +520,7 @@ public class FileLogProjection {
}
/** Metadata of a projected arrow record batch. */
- public static final class ProjectedArrowBatch {
+ static final class ProjectedArrowBatch {
/** Number of records. */
final long numRecords;
diff --git a/fluss-common/src/main/java/org/apache/fluss/record/ProjectionPushdownCache.java b/fluss-common/src/main/java/org/apache/fluss/record/ProjectionPushdownCache.java
new file mode 100644
index 000000000..dc5821e70
--- /dev/null
+++ b/fluss-common/src/main/java/org/apache/fluss/record/ProjectionPushdownCache.java
@@ -0,0 +1,86 @@
+/*
+ * 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.record;
+
+import org.apache.fluss.record.FileLogProjection.ProjectionInfo;
+import org.apache.fluss.shaded.guava32.com.google.common.cache.Cache;
+import org.apache.fluss.shaded.guava32.com.google.common.cache.CacheBuilder;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.ThreadSafe;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Objects;
+
+@ThreadSafe
+public class ProjectionPushdownCache {
+
+ final Cache<ProjectionKey, ProjectionInfo> projectionCache;
+
+ public ProjectionPushdownCache() {
+ // currently, the cache is shared at TabletServer level, so we give a large max size, but
+ // give a short expiration time.
+ // TODO: make the cache parameter configurable
+ this.projectionCache =
+ CacheBuilder.newBuilder()
+ .maximumSize(1000)
+ .expireAfterAccess(Duration.ofMinutes(3))
+ .build();
+ }
+
+ @Nullable
+ public ProjectionInfo getProjectionInfo(long tableId, short schemaId, int[] selectedColumnIds) {
+ ProjectionKey key = new ProjectionKey(tableId, schemaId, selectedColumnIds);
+ return projectionCache.getIfPresent(key);
+ }
+
+ public void setProjectionInfo(
+ long tableId, short schemaId, int[] selectedColumnIds, ProjectionInfo projectionInfo) {
+ ProjectionKey key = new ProjectionKey(tableId, schemaId, selectedColumnIds);
+ projectionCache.put(key, projectionInfo);
+ }
+
+ static final class ProjectionKey {
+ private final long tableId;
+ private final short schemaId;
+ private final int[] selectedColumnIds;
+
+ ProjectionKey(long tableId, short schemaId, int[] selectedColumnIds) {
+ this.tableId = tableId;
+ this.schemaId = schemaId;
+ this.selectedColumnIds = selectedColumnIds;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof ProjectionKey)) {
+ return false;
+ }
+ ProjectionKey that = (ProjectionKey) o;
+ return tableId == that.tableId
+ && schemaId == that.schemaId
+ && Arrays.equals(selectedColumnIds, that.selectedColumnIds);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(tableId, schemaId, Arrays.hashCode(selectedColumnIds));
+ }
+ }
+}
diff --git a/fluss-common/src/test/java/org/apache/fluss/record/FileLogProjectionTest.java b/fluss-common/src/test/java/org/apache/fluss/record/FileLogProjectionTest.java
index 2c3ff36ef..9a27089ac 100644
--- a/fluss-common/src/test/java/org/apache/fluss/record/FileLogProjectionTest.java
+++ b/fluss-common/src/test/java/org/apache/fluss/record/FileLogProjectionTest.java
@@ -78,7 +78,8 @@ class FileLogProjectionTest {
TestData.DATA2,
TestData.DATA2);
- FileLogProjection projection = new FileLogProjection();
+ ProjectionPushdownCache cache = new ProjectionPushdownCache();
+ FileLogProjection projection = new FileLogProjection(cache);
// get schema during running.
doProjection(
1L,
@@ -87,16 +88,13 @@ class FileLogProjectionTest {
recordsOfData2RowType,
new int[] {0, 2},
recordsOfData2RowType.sizeInBytes());
- FileLogProjection.ProjectionInfo info1 = projection.currentProjection;
+ assertThat(cache.projectionCache.size()).isEqualTo(1);
+ FileLogProjection.ProjectionInfo info1 =
+ cache.getProjectionInfo(1L, schemaId, new int[] {0, 2});
assertThat(info1).isNotNull();
assertThat(info1.nodesProjection.stream().toArray()).isEqualTo(new int[] {0, 2});
// a int: [0,1] ; b string: [2,3,4] ; c string: [5,6,7]
assertThat(info1.buffersProjection.stream().toArray()).isEqualTo(new int[] {0, 1, 5, 6, 7});
- assertThat(projection.projectionsCache).hasSize(1);
- assertThat(
- projection.projectionsCache.get(
- new FileLogProjection.ProjectionKey(1L, schemaId)))
- .isSameAs(info1);
doProjection(
2L,
@@ -105,37 +103,13 @@ class FileLogProjectionTest {
recordsOfData2RowType,
new int[] {1},
recordsOfData2RowType.sizeInBytes());
- FileLogProjection.ProjectionInfo info2 = projection.currentProjection;
+ assertThat(cache.projectionCache.size()).isEqualTo(2);
+ FileLogProjection.ProjectionInfo info2 =
+ cache.getProjectionInfo(2L, schemaId, new int[] {1});
assertThat(info2).isNotNull();
assertThat(info2.nodesProjection.stream().toArray()).isEqualTo(new int[] {1});
// a int: [0,1] ; b string: [2,3,4] ; c string: [5,6,7]
assertThat(info2.buffersProjection.stream().toArray()).isEqualTo(new int[] {2, 3, 4});
- assertThat(projection.projectionsCache).hasSize(2);
- assertThat(
- projection.projectionsCache.get(
- new FileLogProjection.ProjectionKey(2L, schemaId)))
- .isSameAs(info2);
-
- doProjection(
- 1L,
- schemaId,
- projection,
- recordsOfData2RowType,
- new int[] {0, 2},
- recordsOfData2RowType.sizeInBytes());
- assertThat(projection.currentProjection).isNotNull().isSameAs(info1);
-
- assertThatThrownBy(
- () ->
- doProjection(
- 1L,
- schemaId,
- projection,
- recordsOfData2RowType,
- new int[] {1},
- recordsOfData2RowType.sizeInBytes()))
- .isInstanceOf(InvalidColumnProjectionException.class)
- .hasMessage("The schema and projection should be identical for the same table id.");
}
@Test
@@ -149,7 +123,7 @@ class FileLogProjectionTest {
TestData.DATA2_ROW_TYPE,
TestData.DATA2,
TestData.DATA2);
- FileLogProjection projection = new FileLogProjection();
+ FileLogProjection projection = new FileLogProjection(new ProjectionPushdownCache());
assertThatThrownBy(
() ->
doProjection(
@@ -160,7 +134,7 @@ class FileLogProjectionTest {
new int[] {3},
recordsOfData2RowType.sizeInBytes()))
.isInstanceOf(InvalidColumnProjectionException.class)
- .hasMessage("Projected field id 3 is not contains in [0, 1, 2]");
+ .hasMessage("Projected field id 3 is not contained in [0, 1, 2]");
assertThatThrownBy(
() ->
@@ -220,7 +194,7 @@ class FileLogProjectionTest {
doProjection(
1L,
schemaId,
- new FileLogProjection(),
+ new FileLogProjection(new ProjectionPushdownCache()),
fileLogRecords,
projectedFields,
Integer.MAX_VALUE);
@@ -248,7 +222,7 @@ class FileLogProjectionTest {
TestData.DATA1_ROW_TYPE,
TestData.DATA1,
TestData.ANOTHER_DATA1);
- FileLogProjection projection = new FileLogProjection();
+ FileLogProjection projection = new FileLogProjection(new ProjectionPushdownCache());
// overwrite the wrong decoding byte order endian
projection.getLogHeaderBuffer().order(ByteOrder.BIG_ENDIAN);
assertThatThrownBy(
@@ -305,7 +279,10 @@ class FileLogProjectionTest {
int maxBytes = totalSize / i;
List<Object[]> results =
doProjection(
- new FileLogProjection(), fileLogRecords, new int[] {0, 1}, maxBytes);
+ new FileLogProjection(new ProjectionPushdownCache()),
+ fileLogRecords,
+ new int[] {0, 1},
+ maxBytes);
if (results.isEmpty()) {
hasEmpty = true;
} else if (results.size() == TestData.DATA1.size()) {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/FetchParams.java b/fluss-server/src/main/java/org/apache/fluss/server/log/FetchParams.java
index 847bfa883..b08ad67f8 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/log/FetchParams.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/log/FetchParams.java
@@ -21,6 +21,7 @@ import org.apache.fluss.annotation.VisibleForTesting;
import org.apache.fluss.compression.ArrowCompressionInfo;
import org.apache.fluss.metadata.SchemaGetter;
import org.apache.fluss.record.FileLogProjection;
+import org.apache.fluss.record.ProjectionPushdownCache;
import org.apache.fluss.rpc.messages.FetchLogRequest;
import javax.annotation.Nullable;
@@ -99,13 +100,14 @@ public final class FetchParams {
int maxFetchBytes,
SchemaGetter schemaGetter,
ArrowCompressionInfo compressionInfo,
- @Nullable int[] projectedFields) {
+ @Nullable int[] projectedFields,
+ ProjectionPushdownCache projectionCache) {
this.fetchOffset = fetchOffset;
this.maxFetchBytes = maxFetchBytes;
if (projectedFields != null) {
projectionEnabled = true;
if (fileLogProjection == null) {
- fileLogProjection = new FileLogProjection();
+ fileLogProjection = new FileLogProjection(projectionCache);
}
fileLogProjection.setCurrentProjection(
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
index 9f6e4d8ad..206586bf1 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
@@ -39,6 +39,7 @@ import org.apache.fluss.metrics.MetricNames;
import org.apache.fluss.metrics.groups.MetricGroup;
import org.apache.fluss.record.KvRecordBatch;
import org.apache.fluss.record.MemoryLogRecords;
+import org.apache.fluss.record.ProjectionPushdownCache;
import org.apache.fluss.remote.RemoteLogFetchInfo;
import org.apache.fluss.remote.RemoteLogSegment;
import org.apache.fluss.rpc.RpcClient;
@@ -149,6 +150,7 @@ public class ReplicaManager {
private final Map<TableBucket, HostedReplica> allReplicas = MapUtils.newConcurrentHashMap();
private final TabletServerMetadataCache metadataCache;
+ private final ProjectionPushdownCache projectionsCache = new ProjectionPushdownCache();
private final Lock replicaStateChangeLock = new ReentrantLock();
/**
@@ -1083,7 +1085,8 @@ public class ReplicaManager {
adjustedMaxBytes,
replica.getSchemaGetter(),
replica.getArrowCompressionInfo(),
- fetchReqInfo.getProjectFields());
+ fetchReqInfo.getProjectFields(),
+ projectionsCache);
LogReadInfo readInfo = replica.fetchRecords(fetchParams);
// Once we read from a non-empty bucket, we stop ignoring request and bucket
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java
index 87b768608..f2013a91a 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java
@@ -38,6 +38,7 @@ import org.apache.fluss.record.KvRecordTestUtils;
import org.apache.fluss.record.LogRecords;
import org.apache.fluss.record.LogTestBase;
import org.apache.fluss.record.MemoryLogRecords;
+import org.apache.fluss.record.ProjectionPushdownCache;
import org.apache.fluss.record.TestData;
import org.apache.fluss.record.TestingSchemaGetter;
import org.apache.fluss.record.bytesview.MultiBytesView;
@@ -651,9 +652,10 @@ class KvTabletTest {
TablePath tablePath = TablePath.of("testDb", tableName);
initLogTabletAndKvTablet(tablePath, DATA1_SCHEMA_PK, config);
RowType rowType = DATA1_SCHEMA_PK.getRowType();
+ ProjectionPushdownCache projectionCache = new ProjectionPushdownCache();
FileLogProjection logProjection = null;
if (doProjection) {
- logProjection = new FileLogProjection();
+ logProjection = new FileLogProjection(projectionCache);
logProjection.setCurrentProjection(
0L, schemaGetter, DEFAULT_COMPRESSION, new int[] {0});
}
@@ -749,7 +751,7 @@ class KvTabletTest {
schemaGetter.updateLatestSchemaInfo(new SchemaInfo(DATA2_SCHEMA, 2));
readLogRowType = doProjection ? DATA2_ROW_TYPE.project(new int[] {0}) : DATA2_ROW_TYPE;
if (doProjection) {
- logProjection = new FileLogProjection();
+ logProjection = new FileLogProjection(projectionCache);
logProjection.setCurrentProjection(
0L, schemaGetter, DEFAULT_COMPRESSION, new int[] {0});
}
@@ -834,9 +836,10 @@ class KvTabletTest {
KvRecordTestUtils.KvRecordFactory kvRecordFactory =
KvRecordTestUtils.KvRecordFactory.of(rowType);
+ ProjectionPushdownCache projectionCache = new ProjectionPushdownCache();
FileLogProjection logProjection = null;
if (doProjection) {
- logProjection = new FileLogProjection();
+ logProjection = new FileLogProjection(projectionCache);
logProjection.setCurrentProjection(
0L, schemaGetter, DEFAULT_COMPRESSION, new int[] {0});
}
@@ -956,7 +959,7 @@ class KvTabletTest {
? newSchema.getRowType().project(new int[] {0})
: newSchema.getRowType();
if (doProjection) {
- logProjection = new FileLogProjection();
+ logProjection = new FileLogProjection(projectionCache);
logProjection.setCurrentProjection(
0L, schemaGetter, DEFAULT_COMPRESSION, new int[] {0});
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/FetchParamsTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/FetchParamsTest.java
index 132de4482..d4c2be57b 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/log/FetchParamsTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/log/FetchParamsTest.java
@@ -19,6 +19,7 @@ package org.apache.fluss.server.log;
import org.apache.fluss.metadata.SchemaInfo;
import org.apache.fluss.record.FileLogProjection;
+import org.apache.fluss.record.ProjectionPushdownCache;
import org.apache.fluss.record.TestData;
import org.apache.fluss.record.TestingSchemaGetter;
@@ -33,13 +34,15 @@ class FetchParamsTest {
@Test
void testSetCurrentFetch() {
FetchParams fetchParams = new FetchParams(1, 100);
+ ProjectionPushdownCache projectionCache = new ProjectionPushdownCache();
fetchParams.setCurrentFetch(
1L,
20L,
1024,
new TestingSchemaGetter(new SchemaInfo(TestData.DATA1_SCHEMA, (short) 1)),
DEFAULT_COMPRESSION,
- null);
+ null,
+ projectionCache);
assertThat(fetchParams.fetchOffset()).isEqualTo(20L);
assertThat(fetchParams.maxFetchBytes()).isEqualTo(1024);
assertThat(fetchParams.projection()).isNull();
@@ -50,7 +53,8 @@ class FetchParamsTest {
512,
new TestingSchemaGetter(new SchemaInfo(TestData.DATA2_SCHEMA, (short) 1)),
DEFAULT_COMPRESSION,
- new int[] {0, 2});
+ new int[] {0, 2},
+ projectionCache);
assertThat(fetchParams.fetchOffset()).isEqualTo(30L);
assertThat(fetchParams.maxFetchBytes()).isEqualTo(512);
assertThat(fetchParams.projection()).isNotNull();
@@ -63,7 +67,8 @@ class FetchParamsTest {
256,
new TestingSchemaGetter(new SchemaInfo(TestData.DATA1_SCHEMA, (short) 1)),
DEFAULT_COMPRESSION,
- null);
+ null,
+ projectionCache);
assertThat(fetchParams.projection()).isNull();
fetchParams.setCurrentFetch(
@@ -72,7 +77,8 @@ class FetchParamsTest {
512,
new TestingSchemaGetter(new SchemaInfo(TestData.DATA2_SCHEMA, (short) 1)),
DEFAULT_COMPRESSION,
- new int[] {0, 2});
+ new int[] {0, 2},
+ projectionCache);
// the FileLogProjection should be cached
assertThat(fetchParams.projection()).isNotNull().isSameAs(prevProjection);
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
index e45a5557c..fcb4ec88c 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
@@ -31,6 +31,7 @@ import org.apache.fluss.record.KvRecordTestUtils;
import org.apache.fluss.record.LogRecordBatch;
import org.apache.fluss.record.LogRecords;
import org.apache.fluss.record.MemoryLogRecords;
+import org.apache.fluss.record.ProjectionPushdownCache;
import org.apache.fluss.server.entity.NotifyLeaderAndIsrData;
import org.apache.fluss.server.kv.KvTablet;
import org.apache.fluss.server.kv.snapshot.CompletedSnapshot;
@@ -137,8 +138,15 @@ final class ReplicaTest extends ReplicaTestBase {
(int)
conf.get(ConfigOptions.CLIENT_SCANNER_LOG_FETCH_MAX_BYTES)
.getBytes());
+ ProjectionPushdownCache projectionCache = new ProjectionPushdownCache();
fetchParams.setCurrentFetch(
- DATA1_TABLE_ID, 0, Integer.MAX_VALUE, schemaGetter, DEFAULT_COMPRESSION, null);
+ DATA1_TABLE_ID,
+ 0,
+ Integer.MAX_VALUE,
+ schemaGetter,
+ DEFAULT_COMPRESSION,
+ null,
+ projectionCache);
LogReadInfo logReadInfo = logReplica.fetchRecords(fetchParams);
assertLogRecordsEquals(
DATA1_ROW_TYPE, logReadInfo.getFetchedData().getRecords(), DATA1, schemaGetter);
@@ -163,7 +171,8 @@ final class ReplicaTest extends ReplicaTestBase {
Integer.MAX_VALUE,
schemaGetter,
DEFAULT_COMPRESSION,
- null);
+ null,
+ projectionCache);
logReadInfo = logReplica.fetchRecords(fetchParams);
assertLogRecordsEquals(
2, DATA2_ROW_TYPE, logReadInfo.getFetchedData().getRecords(), DATA2, schemaGetter);
@@ -808,7 +817,8 @@ final class ReplicaTest extends ReplicaTestBase {
Integer.MAX_VALUE,
replica.getSchemaGetter(),
DEFAULT_COMPRESSION,
- null);
+ null,
+ new ProjectionPushdownCache());
LogReadInfo logReadInfo = replica.fetchRecords(fetchParams);
return logReadInfo.getFetchedData().getRecords();
}
(fluss) 06/07: revert FlinkAsFlussRow changes
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch ci-add-column
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit aa31e1c4c8c7d9a6626b792c24ed6cfa32215f6f
Author: Jark Wu <ja...@apache.org>
AuthorDate: Mon Dec 1 01:21:36 2025 +0800
revert FlinkAsFlussRow changes
---
.../apache/fluss/flink/row/FlinkAsFlussRow.java | 100 ++++-----------------
.../sink/serializer/FlussSerializationSchema.java | 7 +-
.../serializer/RowDataSerializationSchema.java | 28 +++++-
.../sink/serializer/SerializerInitContextImpl.java | 2 +-
.../fluss/flink/sink/writer/FlinkSinkWriter.java | 19 +---
5 files changed, 50 insertions(+), 106 deletions(-)
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/row/FlinkAsFlussRow.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/row/FlinkAsFlussRow.java
index 314e4b40f..003e910ea 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/row/FlinkAsFlussRow.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/row/FlinkAsFlussRow.java
@@ -23,34 +23,17 @@ import org.apache.fluss.row.InternalArray;
import org.apache.fluss.row.InternalRow;
import org.apache.fluss.row.TimestampLtz;
import org.apache.fluss.row.TimestampNtz;
-import org.apache.fluss.types.RowType;
import org.apache.flink.table.data.DecimalData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.TimestampData;
-import javax.annotation.Nullable;
-
-import java.util.List;
-
/** Wraps a Flink {@link RowData} as a Fluss {@link InternalRow}. */
public class FlinkAsFlussRow implements InternalRow {
private RowData flinkRow;
- /**
- * Creates a FlinkAsFlussRow with a mapping from Fluss field index to Flink field index. If the
- * indexMapping is null, flink and flink share same field name.
- */
- @Nullable private final int[] indexMapping;
-
- public FlinkAsFlussRow() {
- this(null);
- }
-
- public FlinkAsFlussRow(int[] indexMapping) {
- this.indexMapping = indexMapping;
- }
+ public FlinkAsFlussRow() {}
public FlinkAsFlussRow replace(RowData flinkRow) {
this.flinkRow = flinkRow;
@@ -59,94 +42,62 @@ public class FlinkAsFlussRow implements InternalRow {
@Override
public int getFieldCount() {
- if (indexMapping == null) {
- return flinkRow.getArity();
- } else {
- return indexMapping.length;
- }
+ return flinkRow.getArity();
}
@Override
public boolean isNullAt(int pos) {
- // if no mapping, means that the field is not found in Flink, just ignore it.
- if ((indexMapping != null && indexMapping[pos] == -1)) {
- return true;
- }
-
- // If pos is larger than flinkRow.getArity(), it indicates that the Fluss table's data type
- // is wider than the data provided by Flink.
- // This often occurs when a schema change happens to the catalog table before job restart.
- // Only appending columns at the end is compatible, just need to ignore the trailing
- // columns.
- // Other types of schema changes (e.g., dropping columns) would lead to incompatibility.
- // Therefore, Fluss currently only supports appending columns, and here we simply ignore any
- // extra columns.
- int flussPos = indexMapping == null ? pos : indexMapping[pos];
- if (flussPos >= flinkRow.getArity()) {
- return true;
- }
-
- return flinkRow.isNullAt(flussPos);
+ return flinkRow.isNullAt(pos);
}
@Override
public boolean getBoolean(int pos) {
- int flussPos = indexMapping == null ? pos : indexMapping[pos];
- return flinkRow.getBoolean(flussPos);
+ return flinkRow.getBoolean(pos);
}
@Override
public byte getByte(int pos) {
- int flussPos = indexMapping == null ? pos : indexMapping[pos];
- return flinkRow.getByte(flussPos);
+ return flinkRow.getByte(pos);
}
@Override
public short getShort(int pos) {
- int flussPos = indexMapping == null ? pos : indexMapping[pos];
- return flinkRow.getShort(flussPos);
+ return flinkRow.getShort(pos);
}
@Override
public int getInt(int pos) {
- int flussPos = indexMapping == null ? pos : indexMapping[pos];
- return flinkRow.getInt(flussPos);
+ return flinkRow.getInt(pos);
}
@Override
public long getLong(int pos) {
- int flussPos = indexMapping == null ? pos : indexMapping[pos];
- return flinkRow.getLong(flussPos);
+ return flinkRow.getLong(pos);
}
@Override
public float getFloat(int pos) {
- int flussPos = indexMapping == null ? pos : indexMapping[pos];
- return flinkRow.getFloat(flussPos);
+ return flinkRow.getFloat(pos);
}
@Override
public double getDouble(int pos) {
- int flussPos = indexMapping == null ? pos : indexMapping[pos];
- return flinkRow.getDouble(flussPos);
+ return flinkRow.getDouble(pos);
}
@Override
public BinaryString getChar(int pos, int length) {
- int flussPos = indexMapping == null ? pos : indexMapping[pos];
- return BinaryString.fromBytes(flinkRow.getString(flussPos).toBytes());
+ return BinaryString.fromBytes(flinkRow.getString(pos).toBytes());
}
@Override
public BinaryString getString(int pos) {
- int flussPos = indexMapping == null ? pos : indexMapping[pos];
- return BinaryString.fromBytes(flinkRow.getString(flussPos).toBytes());
+ return BinaryString.fromBytes(flinkRow.getString(pos).toBytes());
}
@Override
public Decimal getDecimal(int pos, int precision, int scale) {
- int flussPos = indexMapping == null ? pos : indexMapping[pos];
- return fromFlinkDecimal(flinkRow.getDecimal(flussPos, precision, scale));
+ return fromFlinkDecimal(flinkRow.getDecimal(pos, precision, scale));
}
public static Decimal fromFlinkDecimal(DecimalData decimal) {
@@ -161,43 +112,26 @@ public class FlinkAsFlussRow implements InternalRow {
@Override
public TimestampNtz getTimestampNtz(int pos, int precision) {
- int flussPos = indexMapping == null ? pos : indexMapping[pos];
- TimestampData timestamp = flinkRow.getTimestamp(flussPos, precision);
+ TimestampData timestamp = flinkRow.getTimestamp(pos, precision);
return TimestampNtz.fromMillis(
timestamp.getMillisecond(), timestamp.getNanoOfMillisecond());
}
@Override
public TimestampLtz getTimestampLtz(int pos, int precision) {
- int flussPos = indexMapping == null ? pos : indexMapping[pos];
- TimestampData timestamp = flinkRow.getTimestamp(flussPos, precision);
+ TimestampData timestamp = flinkRow.getTimestamp(pos, precision);
return TimestampLtz.fromEpochMillis(
timestamp.getMillisecond(), timestamp.getNanoOfMillisecond());
}
@Override
public byte[] getBinary(int pos, int length) {
- int flussPos = indexMapping == null ? pos : indexMapping[pos];
- return flinkRow.getBinary(flussPos);
+ return flinkRow.getBinary(pos);
}
@Override
public byte[] getBytes(int pos) {
- int flussPos = indexMapping == null ? pos : indexMapping[pos];
- return flinkRow.getBinary(flussPos);
- }
-
- /** Converts a Flink {@link RowType} to a Fluss {@link RowType}, based on the field name. */
- public static FlinkAsFlussRow from(
- org.apache.flink.table.types.logical.RowType flinkRowType, RowType flussRowType) {
- int[] indexMapping = new int[flussRowType.getFieldCount()];
- List<String> fieldNames = flussRowType.getFieldNames();
- for (int i = 0; i < flussRowType.getFieldCount(); i++) {
- String fieldName = fieldNames.get(i);
- int fieldIndex = flinkRowType.getFieldIndex(fieldName);
- indexMapping[i] = fieldIndex;
- }
- return new FlinkAsFlussRow(indexMapping);
+ return flinkRow.getBinary(pos);
}
@Override
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/serializer/FlussSerializationSchema.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/serializer/FlussSerializationSchema.java
index 0f3fff210..2096b1095 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/serializer/FlussSerializationSchema.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/serializer/FlussSerializationSchema.java
@@ -69,10 +69,11 @@ public interface FlussSerializationSchema<T> extends Serializable {
RowType getRowSchema();
/**
- * Returns the Flink table row schema.
+ * Returns the input row schema which is the Flink {@link
+ * org.apache.flink.table.data.RowData}.
*
- * @return The schema of the Flink table row.
+ * @return The schema of the input Flink row.
*/
- org.apache.flink.table.types.logical.RowType getTableRowType();
+ org.apache.flink.table.types.logical.RowType getInputRowSchema();
}
}
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/serializer/RowDataSerializationSchema.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/serializer/RowDataSerializationSchema.java
index 05e2921fc..440c70485 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/serializer/RowDataSerializationSchema.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/serializer/RowDataSerializationSchema.java
@@ -21,10 +21,15 @@ import org.apache.fluss.flink.row.FlinkAsFlussRow;
import org.apache.fluss.flink.row.OperationType;
import org.apache.fluss.flink.row.RowWithOp;
import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.ProjectedRow;
import org.apache.flink.table.data.RowData;
import org.apache.flink.types.RowKind;
+import javax.annotation.Nullable;
+
+import java.util.List;
+
/** Default implementation of RowDataConverter for RowData. */
public class RowDataSerializationSchema implements FlussSerializationSchema<RowData> {
private static final long serialVersionUID = 1L;
@@ -44,6 +49,12 @@ public class RowDataSerializationSchema implements FlussSerializationSchema<RowD
*/
private transient FlinkAsFlussRow converter;
+ /**
+ * The projected row for output, used when schema evolution occurs, because we want to write
+ * rows using new schema (e.g., fill null for new columns).
+ */
+ @Nullable private transient ProjectedRow outputProjection;
+
/**
* Constructs a new {@code RowSerializationSchema}.
*
@@ -63,7 +74,19 @@ public class RowDataSerializationSchema implements FlussSerializationSchema<RowD
*/
@Override
public void open(InitializationContext context) throws Exception {
- this.converter = FlinkAsFlussRow.from(context.getTableRowType(), context.getRowSchema());
+ this.converter = new FlinkAsFlussRow();
+ List<String> targetFieldNames = context.getRowSchema().getFieldNames();
+ List<String> inputFieldNames = context.getInputRowSchema().getFieldNames();
+ if (targetFieldNames.size() != inputFieldNames.size()) {
+ // there is a schema evolution happens (e.g., ADD COLUMN), need to build index mapping
+ int[] indexMapping = new int[targetFieldNames.size()];
+ for (int i = 0; i < targetFieldNames.size(); i++) {
+ String fieldName = targetFieldNames.get(i);
+ int fieldIndex = inputFieldNames.indexOf(fieldName);
+ indexMapping[i] = fieldIndex;
+ }
+ outputProjection = ProjectedRow.from(indexMapping);
+ }
}
/**
@@ -82,6 +105,9 @@ public class RowDataSerializationSchema implements FlussSerializationSchema<RowD
"Converter not initialized. The open() method must be called before serializing records.");
}
InternalRow row = converter.replace(value);
+ if (outputProjection != null) {
+ row = outputProjection.replaceRow(row);
+ }
OperationType opType = toOperationType(value.getRowKind());
return new RowWithOp(row, opType);
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/serializer/SerializerInitContextImpl.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/serializer/SerializerInitContextImpl.java
index 93b849e58..cc1d168cd 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/serializer/SerializerInitContextImpl.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/serializer/SerializerInitContextImpl.java
@@ -47,7 +47,7 @@ public class SerializerInitContextImpl implements FlussSerializationSchema.Initi
}
@Override
- public org.apache.flink.table.types.logical.RowType getTableRowType() {
+ public org.apache.flink.table.types.logical.RowType getInputRowSchema() {
return flinkRowType;
}
}
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/FlinkSinkWriter.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/FlinkSinkWriter.java
index a4469fd32..97a70ce25 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/FlinkSinkWriter.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/FlinkSinkWriter.java
@@ -29,7 +29,6 @@ import org.apache.fluss.flink.row.OperationType;
import org.apache.fluss.flink.row.RowWithOp;
import org.apache.fluss.flink.sink.serializer.FlussSerializationSchema;
import org.apache.fluss.flink.sink.serializer.SerializerInitContextImpl;
-import org.apache.fluss.flink.utils.FlinkConversions;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.metrics.Gauge;
@@ -114,7 +113,7 @@ public abstract class FlinkSinkWriter<InputT> implements SinkWriter<InputT> {
"Current Fluss Schema is {}, Table RowType is {}",
table.getTableInfo().getSchema(),
tableRowType);
- // sanityCheck(table.getTableInfo());
+ sanityCheck(table.getTableInfo());
try {
this.serializationSchema.open(
@@ -209,22 +208,6 @@ public abstract class FlinkSinkWriter<InputT> implements SinkWriter<InputT> {
"Primary key constraint is not matched between metadata in Fluss (%s) and Flink (%s).",
flussTableInfo.hasPrimaryKey(), hasPrimaryKey));
}
- RowType currentTableRowType = FlinkConversions.toFlinkRowType(flussTableInfo.getRowType());
- if (!this.tableRowType.copy(false).equals(currentTableRowType.copy(false))) {
- // The default nullability of Flink row type and Fluss row type might be not the same,
- // thus we need to compare the row type without nullability here.
-
- // Throw exception if the schema is the not same, this should rarely happen because we
- // only allow fluss tables derived from fluss catalog. But this can happen if an ALTER
- // TABLE command executed on the fluss table, after the job is submitted but before the
- // SinkFunction is opened.
- throw new ValidationException(
- "The Flink query schema is not matched to current Fluss table schema. "
- + "\nFlink query schema: "
- + this.tableRowType
- + "\nFluss table schema: "
- + currentTableRowType);
- }
}
private long computeSendTime() {
(fluss) 05/07: WIP2
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch ci-add-column
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 5d65a20b79fbefaae07955ffc296267dbde39b43
Author: Jark Wu <ja...@apache.org>
AuthorDate: Mon Dec 1 01:01:42 2025 +0800
WIP2
---
.../src/main/java/org/apache/fluss/metadata/Schema.java | 14 ++++++--------
.../java/org/apache/fluss/record/KvRecordReadContext.java | 1 +
.../java/org/apache/fluss/record/LogRecordReadContext.java | 1 +
.../src/main/java/org/apache/fluss/row/ProjectedRow.java | 6 +++---
.../java/org/apache/fluss/row/encode/ValueDecoder.java | 1 +
.../java/org/apache/fluss/row/encode/ValueEncoder.java | 1 +
.../src/test/java/org/apache/fluss/record/LogTestBase.java | 1 -
.../test/java/org/apache/fluss/row/ProjectedRowTest.java | 1 +
.../org/apache/fluss/flink/row/FlinkAsFlussRowTest.java | 3 +--
9 files changed, 15 insertions(+), 14 deletions(-)
diff --git a/fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java b/fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java
index c065a3711..0ae0c8b97 100644
--- a/fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java
+++ b/fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java
@@ -155,6 +155,7 @@ public final class Schema implements Serializable {
return keyIndexes;
}
+ /** Returns the highest field ID in this schema. */
public int getHighestFieldId() {
return highestFieldId;
}
@@ -250,24 +251,21 @@ public final class Schema implements Serializable {
/** Adopts all columns from the given list. */
public Builder fromColumns(List<Column> inputColumns) {
- boolean nonMatchColumnId =
+ boolean nonSetColumnId =
inputColumns.stream()
.noneMatch(column -> column.columnId != Column.UNKNOWN_COLUMN_ID);
- boolean allMatchColumnId =
+ boolean allSetColumnId =
inputColumns.stream()
.allMatch(column -> column.columnId != Column.UNKNOWN_COLUMN_ID);
checkState(
- nonMatchColumnId || allMatchColumnId,
+ nonSetColumnId || allSetColumnId,
"All columns must have columnId or none of them must have columnId.");
- if (allMatchColumnId) {
+ if (allSetColumnId) {
columns.addAll(inputColumns);
highestFieldId =
new AtomicInteger(
- inputColumns.stream()
- .mapToInt(Column::getColumnId)
- .max()
- .orElse(-1));
+ columns.stream().mapToInt(Column::getColumnId).max().orElse(-1));
} else {
// if all columnId is not set, this maybe from old version schema. Just use its
// position as columnId.
diff --git a/fluss-common/src/main/java/org/apache/fluss/record/KvRecordReadContext.java b/fluss-common/src/main/java/org/apache/fluss/record/KvRecordReadContext.java
index 5fffc3f4f..195312cef 100644
--- a/fluss-common/src/main/java/org/apache/fluss/record/KvRecordReadContext.java
+++ b/fluss-common/src/main/java/org/apache/fluss/record/KvRecordReadContext.java
@@ -31,6 +31,7 @@ public class KvRecordReadContext implements KvRecordBatch.ReadContext {
private final KvFormat kvFormat;
private final SchemaGetter schemaGetter;
+ // TODO reuse
private final Map<Integer, RowDecoder> rowDecoderCache;
private KvRecordReadContext(KvFormat kvFormat, SchemaGetter schemaGetter) {
diff --git a/fluss-common/src/main/java/org/apache/fluss/record/LogRecordReadContext.java b/fluss-common/src/main/java/org/apache/fluss/record/LogRecordReadContext.java
index 571e75221..5ec5c7c77 100644
--- a/fluss-common/src/main/java/org/apache/fluss/record/LogRecordReadContext.java
+++ b/fluss-common/src/main/java/org/apache/fluss/record/LogRecordReadContext.java
@@ -244,6 +244,7 @@ public class LogRecordReadContext implements LogRecordBatch.ReadContext, AutoClo
if (isSameRowType(schemaId)) {
return null;
}
+ // TODO: should we cache the projection?
Schema originSchema = schemaGetter.getSchema(schemaId);
Schema expectedSchema = schemaGetter.getSchema(targetSchemaId);
return ProjectedRow.from(originSchema, expectedSchema);
diff --git a/fluss-common/src/main/java/org/apache/fluss/row/ProjectedRow.java b/fluss-common/src/main/java/org/apache/fluss/row/ProjectedRow.java
index 6b4c516d1..d793f342a 100644
--- a/fluss-common/src/main/java/org/apache/fluss/row/ProjectedRow.java
+++ b/fluss-common/src/main/java/org/apache/fluss/row/ProjectedRow.java
@@ -36,11 +36,11 @@ import static org.apache.fluss.utils.SchemaUtil.getIndexMapping;
public class ProjectedRow implements InternalRow {
public static final int UNEXIST_MAPPING = -1;
- protected final int[] indexMapping;
+ private final int[] indexMapping;
- protected InternalRow row;
+ private InternalRow row;
- protected ProjectedRow(int[] indexMapping) {
+ private ProjectedRow(int[] indexMapping) {
this.indexMapping = indexMapping;
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueDecoder.java b/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueDecoder.java
index e0252ae11..5f90ddaef 100644
--- a/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueDecoder.java
+++ b/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueDecoder.java
@@ -36,6 +36,7 @@ import static org.apache.fluss.utils.MapUtils.newConcurrentHashMap;
*/
public class ValueDecoder {
+ // TODO: reuse?
private final Map<Short, RowDecoder> rowDecoders;
private final SchemaGetter schemaGetter;
private final KvFormat kvFormat;
diff --git a/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueEncoder.java b/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueEncoder.java
index 415fac0c2..74eb10094 100644
--- a/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueEncoder.java
+++ b/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueEncoder.java
@@ -42,6 +42,7 @@ public class ValueEncoder {
return values;
}
+ // TODO: ???
public static byte[] encodeRow(
short schemaId, KvFormat kvFormat, RowType currentRowType, InternalRow row)
throws Exception {
diff --git a/fluss-common/src/test/java/org/apache/fluss/record/LogTestBase.java b/fluss-common/src/test/java/org/apache/fluss/record/LogTestBase.java
index 108ab2acf..36aef5b9d 100644
--- a/fluss-common/src/test/java/org/apache/fluss/record/LogTestBase.java
+++ b/fluss-common/src/test/java/org/apache/fluss/record/LogTestBase.java
@@ -72,7 +72,6 @@ public abstract class LogTestBase {
assertLogRecordsListEquals(expected, actual, baseRowType);
}
- // todo: 在这个基础上可以加更多最简单的schema evolution测试，非常好
protected void assertIndexedLogRecordBatchAndRowEquals(
LogRecordBatch actual,
LogRecordBatch expected,
diff --git a/fluss-common/src/test/java/org/apache/fluss/row/ProjectedRowTest.java b/fluss-common/src/test/java/org/apache/fluss/row/ProjectedRowTest.java
index 09467d31a..80b21a719 100644
--- a/fluss-common/src/test/java/org/apache/fluss/row/ProjectedRowTest.java
+++ b/fluss-common/src/test/java/org/apache/fluss/row/ProjectedRowTest.java
@@ -183,6 +183,7 @@ class ProjectedRowTest {
.isExactlyInstanceOf(SchemaChangeException.class)
.hasMessage(
"Expected datatype of column(id=0,name=a) is [INT], while the actual datatype is [BIGINT]");
+
assertThatThrownBy(
() ->
ProjectedRow.from(
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/row/FlinkAsFlussRowTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/row/FlinkAsFlussRowTest.java
index 1f147751a..73ba044d8 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/row/FlinkAsFlussRowTest.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/row/FlinkAsFlussRowTest.java
@@ -31,7 +31,6 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.time.Instant;
-import java.util.stream.IntStream;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.AssertionsForClassTypes.offset;
@@ -60,7 +59,7 @@ class FlinkAsFlussRowTest {
TimestampData.fromEpochMillis(1672531200000L, 3),
new byte[] {1, 2, 3},
null);
- row = new FlinkAsFlussRow(IntStream.range(0, 14).toArray()).replace(flinkRow);
+ row = new FlinkAsFlussRow().replace(flinkRow);
}
@Test
(fluss) 01/07: WIP
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch ci-add-column
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit e715276f23a517c8640a267c85d0aba0e040c874
Author: Jark Wu <ja...@apache.org>
AuthorDate: Sun Nov 30 21:14:28 2025 +0800
WIP
---
.../org/apache/fluss/client/table/FlussTable.java | 6 +---
.../client/table/scanner/batch/BatchScanUtils.java | 3 +-
.../table/scanner/batch/LimitBatchScanner.java | 22 ++++++++++---
.../table/scanner/batch/SnapshotFilesReader.java | 19 +++++++++--
.../client/table/scanner/log/BucketScanStatus.java | 3 +-
.../fluss/client/table/scanner/log/LogFetcher.java | 3 +-
.../client/table/writer/AbstractTableWriter.java | 2 --
.../fluss/client/write/RecordAccumulator.java | 3 +-
.../java/org/apache/fluss/client/write/Sender.java | 6 ++--
.../fluss/client/admin/FlussAdminITCase.java | 37 ++++++++++++++++++++--
.../batch/KvSnapshotBatchScannerITCase.java | 4 +--
.../table/scanner/log/LogFetchCollectorTest.java | 3 +-
.../client/table/scanner/log/LogFetcherTest.java | 1 +
.../fluss/client/write/RecordAccumulatorTest.java | 1 -
.../org/apache/fluss/exception/ApiException.java | 16 +++++++++-
.../fluss/exception/SchemaChangeException.java | 4 +++
.../java/org/apache/fluss/metadata/Schema.java | 1 +
.../org/apache/fluss/metadata/TableChange.java | 2 +-
.../java/org/apache/fluss/metadata/TableInfo.java | 5 +++
.../java/org/apache/fluss/utils/Projection.java | 6 +---
.../java/org/apache/fluss/utils/SchemaUtil.java | 21 ++++++++----
.../server/metadata/SchemaMetadataManager.java | 1 +
22 files changed, 122 insertions(+), 47 deletions(-)
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/FlussTable.java b/fluss-client/src/main/java/org/apache/fluss/client/table/FlussTable.java
index 629abe6ab..8532f2a85 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/table/FlussTable.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/table/FlussTable.java
@@ -29,7 +29,6 @@ import org.apache.fluss.client.table.writer.TableAppend;
import org.apache.fluss.client.table.writer.TableUpsert;
import org.apache.fluss.client.table.writer.Upsert;
import org.apache.fluss.metadata.SchemaGetter;
-import org.apache.fluss.metadata.SchemaInfo;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
@@ -55,10 +54,7 @@ public class FlussTable implements Table {
this.tableInfo = tableInfo;
this.hasPrimaryKey = tableInfo.hasPrimaryKey();
this.schemaGetter =
- new ClientSchemaGetter(
- tableInfo.getTablePath(),
- new SchemaInfo(tableInfo.getSchema(), tableInfo.getSchemaId()),
- conn.getAdmin());
+ new ClientSchemaGetter(tablePath, tableInfo.getSchemaInfo(), conn.getAdmin());
}
@Override
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/BatchScanUtils.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/BatchScanUtils.java
index f4e150f30..d36bf9a7d 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/BatchScanUtils.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/BatchScanUtils.java
@@ -71,8 +71,7 @@ public class BatchScanUtils {
// If the scanner has more data, add it back to the queue
scannerQueue.add(scanner);
} else {
- // Close the scanner if it has no more data, and not add it back to the
- // queue
+ // Close the scanner if it has no more data, and not add it back to the queue
scanner.close();
}
} catch (Exception e) {
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/LimitBatchScanner.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/LimitBatchScanner.java
index e6208f368..06c79fce8 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/LimitBatchScanner.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/LimitBatchScanner.java
@@ -39,6 +39,7 @@ import org.apache.fluss.rpc.messages.LimitScanRequest;
import org.apache.fluss.rpc.messages.LimitScanResponse;
import org.apache.fluss.types.RowType;
import org.apache.fluss.utils.CloseableIterator;
+import org.apache.fluss.utils.SchemaUtil;
import javax.annotation.Nullable;
@@ -47,7 +48,9 @@ import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -64,6 +67,12 @@ public class LimitBatchScanner implements BatchScanner {
private final KvFormat kvFormat;
private final int targetSchemaId;
+ /**
+ * A cache for schema projection mapping from source schema to target. Use HashMap here, because
+ * LimitBatchScanner is used in single thread only.
+ */
+ private final Map<Short, int[]> schemaProjectionCache = new HashMap<>();
+
private boolean endOfInput;
public LimitBatchScanner(
@@ -142,11 +151,14 @@ public class LimitBatchScanner implements BatchScanner {
for (ValueRecord record : valueRecords.records(readContext)) {
InternalRow row = record.getRow();
if (targetSchemaId != record.schemaId()) {
- row =
- ProjectedRow.from(
- schemaGetter.getSchema(record.schemaId()),
- schemaGetter.getSchema(targetSchemaId))
- .replaceRow(row);
+ int[] indexMapping =
+ schemaProjectionCache.computeIfAbsent(
+ record.schemaId(),
+ sourceSchemaId ->
+ SchemaUtil.getIndexMapping(
+ schemaGetter.getSchema(sourceSchemaId),
+ schemaGetter.getSchema(targetSchemaId)));
+ row = ProjectedRow.from(indexMapping).replaceRow(row);
}
scanRows.add(maybeProject(row));
}
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/SnapshotFilesReader.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/SnapshotFilesReader.java
index 10e773609..a7c5e2969 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/SnapshotFilesReader.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/SnapshotFilesReader.java
@@ -30,6 +30,7 @@ import org.apache.fluss.row.encode.ValueDecoder;
import org.apache.fluss.utils.CloseableIterator;
import org.apache.fluss.utils.CloseableRegistry;
import org.apache.fluss.utils.IOUtils;
+import org.apache.fluss.utils.SchemaUtil;
import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.DBOptions;
@@ -43,6 +44,8 @@ import javax.annotation.concurrent.NotThreadSafe;
import java.io.IOException;
import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Map;
/**
* A reader to read kv snapshot files to {@link ScanRecord}s. It will return the {@link ScanRecord}s
@@ -58,6 +61,12 @@ class SnapshotFilesReader implements CloseableIterator<InternalRow> {
@Nullable private final int[] projectedFields;
private RocksIteratorWrapper rocksIteratorWrapper;
+ /**
+ * A cache for schema projection mapping from source schema to target. Use HashMap here, because
+ * SnapshotFilesReader is used in single thread only.
+ */
+ private final Map<Short, int[]> schemaProjectionCache = new HashMap<>();
+
private Snapshot snapshot;
private RocksDBHandle rocksDBHandle;
private boolean isClose = false;
@@ -158,9 +167,13 @@ class SnapshotFilesReader implements CloseableIterator<InternalRow> {
ValueDecoder.Value originValue = valueDecoder.decodeValue(value);
InternalRow originRow = originValue.row;
if (targetSchemaId != originValue.schemaId) {
- originRow =
- ProjectedRow.from(schemaGetter.getSchema(originValue.schemaId), targetSchema)
- .replaceRow(originRow);
+ int[] indexMapping =
+ schemaProjectionCache.computeIfAbsent(
+ originValue.schemaId,
+ sourceSchemaId ->
+ SchemaUtil.getIndexMapping(
+ schemaGetter.getSchema(sourceSchemaId), targetSchema));
+ originRow = ProjectedRow.from(indexMapping).replaceRow(originRow);
}
if (projectedFields != null) {
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/BucketScanStatus.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/BucketScanStatus.java
index bee0910e1..738e04a49 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/BucketScanStatus.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/BucketScanStatus.java
@@ -24,8 +24,7 @@ import org.apache.fluss.annotation.Internal;
class BucketScanStatus {
private long offset; // last consumed position
private long highWatermark; // the high watermark from last fetch
- // TODO add
- // resetStrategy and nextAllowedRetryTimeMs.
+ // TODO add resetStrategy and nextAllowedRetryTimeMs.
public BucketScanStatus() {
this.offset = 0L;
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java
index e8067dd4c..b19347772 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java
@@ -364,8 +364,7 @@ public class LogFetcher implements Closeable {
if (!MemoryLogRecords.EMPTY.equals(logRecords)
|| fetchResultForBucket.getErrorCode() != Errors.NONE.code()) {
// In oder to not signal notEmptyCondition, add completed
- // fetch to
- // buffer until log records is not empty.
+ // fetch to buffer until log records is not empty.
DefaultCompletedFetch completedFetch =
new DefaultCompletedFetch(
tb,
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/writer/AbstractTableWriter.java b/fluss-client/src/main/java/org/apache/fluss/client/table/writer/AbstractTableWriter.java
index f1811dfa2..98a93fdd2 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/table/writer/AbstractTableWriter.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/table/writer/AbstractTableWriter.java
@@ -37,13 +37,11 @@ public abstract class AbstractTableWriter implements TableWriter {
protected final TablePath tablePath;
protected final WriterClient writerClient;
protected final int fieldCount;
- protected TableInfo tableInfo;
private final @Nullable PartitionGetter partitionFieldGetter;
protected AbstractTableWriter(
TablePath tablePath, TableInfo tableInfo, WriterClient writerClient) {
this.tablePath = tablePath;
- this.tableInfo = tableInfo;
this.writerClient = writerClient;
this.fieldCount = tableInfo.getRowType().getFieldCount();
this.partitionFieldGetter =
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/write/RecordAccumulator.java b/fluss-client/src/main/java/org/apache/fluss/client/write/RecordAccumulator.java
index f7d7d3ef0..3a282ba92 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/write/RecordAccumulator.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/write/RecordAccumulator.java
@@ -868,8 +868,7 @@ public final class RecordAccumulator {
// Either we have reached a point where there are batches without a sequence (i.e. never
// been drained and are hence in order by default), or the batch at the front of the
// queue has a sequence greater than the incoming batch. This is the right place to
- // add
- // the incoming batch.
+ // add the incoming batch.
deque.addFirst(batch);
// Now we have to re-insert the previously queued batches in the right order.
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/write/Sender.java b/fluss-client/src/main/java/org/apache/fluss/client/write/Sender.java
index 1b9c79397..e9c285306 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/write/Sender.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/write/Sender.java
@@ -130,7 +130,7 @@ public class Sender implements Runnable {
this.idempotenceManager = idempotenceManager;
this.writerMetricGroup = writerMetricGroup;
- // TODOadd retry logic while send failed. See FLUSS-56364375
+ // TODO add retry logic while send failed. See FLUSS-56364375
}
@VisibleForTesting
@@ -167,7 +167,7 @@ public class Sender implements Runnable {
}
}
- // TODO if force close failed,add logic to abort incomplete batches.
+ // TODO if force close failed, add logic to abort incomplete batches.
LOG.debug("Shutdown of Fluss write sender I/O thread has completed.");
}
@@ -245,7 +245,7 @@ public class Sender implements Runnable {
if (!batches.isEmpty()) {
addToInflightBatches(batches);
- // TODOadd logic for batch expire.
+ // TODO add logic for batch expire.
sendWriteRequests(batches);
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
index 05f196fdd..bd7954284 100644
--- a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
+++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
@@ -354,13 +354,43 @@ class FlussAdminITCase extends ClientToServerITCaseBase {
.get())
.hasMessageContaining("Column c1 must be nullable");
+ assertThatThrownBy(
+ () ->
+ admin.alterTable(
+ tablePath,
+ Collections.singletonList(
+ TableChange.addColumn(
+ "c1",
+ DataTypes.STRING().copy(false),
+ null,
+ TableChange.ColumnPosition
+ .first())),
+ false)
+ .get())
+ .hasMessageContaining("Unsupported ColumnPositionType: FIRST");
+
+ assertThatThrownBy(
+ () ->
+ admin.alterTable(
+ tablePath,
+ Collections.singletonList(
+ TableChange.addColumn(
+ "c1",
+ DataTypes.STRING().copy(false),
+ null,
+ TableChange.ColumnPosition.after(
+ "name"))),
+ false)
+ .get())
+ .hasMessageContaining("Unsupported ColumnPositionType: AFTER(name)");
+
admin.alterTable(
tablePath,
Collections.singletonList(
TableChange.addColumn(
"c1",
DataTypes.STRING(),
- null,
+ "new column c1",
TableChange.ColumnPosition.last())),
false)
.get();
@@ -380,7 +410,10 @@ class FlussAdminITCase extends ClientToServerITCaseBase {
new Schema.Column(
"age", DataTypes.INT(), "person age", (short) 2),
new Schema.Column(
- "c1", DataTypes.STRING(), null, (short) 3)))
+ "c1",
+ DataTypes.STRING(),
+ "new column c1",
+ (short) 3)))
.build();
SchemaInfo schemaInfo = admin.getTableSchema(tablePath).get();
assertThat(schemaInfo).isEqualTo(new SchemaInfo(expectedSchema, 2));
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/batch/KvSnapshotBatchScannerITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/batch/KvSnapshotBatchScannerITCase.java
index 527994777..62b5bdc13 100644
--- a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/batch/KvSnapshotBatchScannerITCase.java
+++ b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/batch/KvSnapshotBatchScannerITCase.java
@@ -134,7 +134,7 @@ class KvSnapshotBatchScannerITCase extends ClientToServerITCaseBase {
Collections.singletonList(
TableChange.addColumn(
"new_column",
- DataTypes.BIGINT().copy(false).copy(true),
+ DataTypes.BIGINT().copy(true),
null,
TableChange.ColumnPosition.last())),
false)
@@ -148,7 +148,7 @@ class KvSnapshotBatchScannerITCase extends ClientToServerITCaseBase {
.column("name", DataTypes.STRING())
.column("new_column", DataTypes.BIGINT())
.build();
- // put into values with old schema.
+ // put into values with new schema.
List<InternalRow> rows = new ArrayList<>();
for (int i = 10; i < 20; i++) {
InternalRow row =
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchCollectorTest.java b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchCollectorTest.java
index 83878a816..99a108e2f 100644
--- a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchCollectorTest.java
+++ b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchCollectorTest.java
@@ -93,8 +93,7 @@ public class LogFetchCollectorTest {
logFetchBuffer.add(completedFetch);
assertThat(logFetchBuffer.isEmpty()).isFalse();
- // Validate that the completed fetch isn't initialized just because we add it to the
- // buffer.
+ // Validate that the completed fetch isn't initialized just because we add it to the buffer
assertThat(completedFetch.isInitialized()).isFalse();
// Fetch the data and validate that we get all the records we want back.
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherTest.java b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherTest.java
index 03f908733..e5a8c8ff0 100644
--- a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherTest.java
+++ b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherTest.java
@@ -199,6 +199,7 @@ public class LogFetcherTest extends ClientToServerITCaseBase {
scanRecords = records.get(tb0);
assertThat(scanRecords.stream().map(ScanRecord::getRow).collect(Collectors.toList()))
.isEqualTo(expectedRows);
+ newSchemaLogFetcher.close();
}
@Test
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/write/RecordAccumulatorTest.java b/fluss-client/src/test/java/org/apache/fluss/client/write/RecordAccumulatorTest.java
index bc4e53b50..47a9c2414 100644
--- a/fluss-client/src/test/java/org/apache/fluss/client/write/RecordAccumulatorTest.java
+++ b/fluss-client/src/test/java/org/apache/fluss/client/write/RecordAccumulatorTest.java
@@ -137,7 +137,6 @@ class RecordAccumulatorTest {
conf = new Configuration();
// init cluster.
cluster = updateCluster(Arrays.asList(bucket1, bucket2, bucket3));
- // todo: 从cluster中获取schema getter
schemaGetter = new TestingSchemaGetter(new SchemaInfo(DATA1_SCHEMA, (short) 1));
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/exception/ApiException.java b/fluss-common/src/main/java/org/apache/fluss/exception/ApiException.java
index 14b0b7f96..b577b8633 100644
--- a/fluss-common/src/main/java/org/apache/fluss/exception/ApiException.java
+++ b/fluss-common/src/main/java/org/apache/fluss/exception/ApiException.java
@@ -25,21 +25,35 @@ public class ApiException extends FlussRuntimeException {
private static final long serialVersionUID = 1L;
+ private final boolean stackTraceEnabled;
+
+ public ApiException(String message, Throwable cause, boolean stackTraceEnabled) {
+ super(message, cause);
+ this.stackTraceEnabled = stackTraceEnabled;
+ }
+
public ApiException(String message, Throwable cause) {
super(message, cause);
+ this.stackTraceEnabled = false;
}
public ApiException(String message) {
super(message);
+ this.stackTraceEnabled = false;
}
public ApiException(Throwable cause) {
super(cause);
+ this.stackTraceEnabled = false;
}
/* avoid the expensive and useless stack trace for api exceptions */
@Override
public Throwable fillInStackTrace() {
- return this;
+ if (stackTraceEnabled) {
+ return super.fillInStackTrace();
+ } else {
+ return this;
+ }
}
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/exception/SchemaChangeException.java b/fluss-common/src/main/java/org/apache/fluss/exception/SchemaChangeException.java
index 8c9446682..77983e441 100644
--- a/fluss-common/src/main/java/org/apache/fluss/exception/SchemaChangeException.java
+++ b/fluss-common/src/main/java/org/apache/fluss/exception/SchemaChangeException.java
@@ -27,4 +27,8 @@ public class SchemaChangeException extends ApiException {
public SchemaChangeException(String message) {
this(message, null);
}
+
+ public SchemaChangeException(String message, boolean stackTraceEnabled) {
+ super(message, null, stackTraceEnabled);
+ }
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java b/fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java
index 367013363..c065a3711 100644
--- a/fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java
+++ b/fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java
@@ -132,6 +132,7 @@ public final class Schema implements Serializable {
return columns.stream().map(Column::getName).collect(Collectors.toList());
}
+ /** Returns all column ids for top-level columns, the nested field ids are not included. */
public List<Integer> getColumnIds() {
return columns.stream().map(Column::getColumnId).collect(Collectors.toList());
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/metadata/TableChange.java b/fluss-common/src/main/java/org/apache/fluss/metadata/TableChange.java
index 36162f30c..e012a38f6 100644
--- a/fluss-common/src/main/java/org/apache/fluss/metadata/TableChange.java
+++ b/fluss-common/src/main/java/org/apache/fluss/metadata/TableChange.java
@@ -402,7 +402,7 @@ public interface TableChange {
@Override
public String toString() {
- return String.format("AFTER %s", columnName);
+ return String.format("AFTER(%s)", columnName);
}
}
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/metadata/TableInfo.java b/fluss-common/src/main/java/org/apache/fluss/metadata/TableInfo.java
index af293f120..eba41e078 100644
--- a/fluss-common/src/main/java/org/apache/fluss/metadata/TableInfo.java
+++ b/fluss-common/src/main/java/org/apache/fluss/metadata/TableInfo.java
@@ -137,6 +137,11 @@ public final class TableInfo {
return schema;
}
+ /** Returns the schema info of the table, including schema and schema id. */
+ public SchemaInfo getSchemaInfo() {
+ return new SchemaInfo(schema, schemaId);
+ }
+
/**
* Returns the row type of the table. The row type is the schema of the table, which defines the
* columns and types of the table.
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/Projection.java b/fluss-common/src/main/java/org/apache/fluss/utils/Projection.java
index c597caf81..7c392b055 100644
--- a/fluss-common/src/main/java/org/apache/fluss/utils/Projection.java
+++ b/fluss-common/src/main/java/org/apache/fluss/utils/Projection.java
@@ -112,11 +112,7 @@ public class Projection {
return projectionPositions;
}
- /**
- * The id of the projection, which is used for fluss server scan.
- *
- * @return
- */
+ /** The id of the projection, which is used for fluss server scan. */
public int[] getProjectionIdInOrder() {
return projectionIdsInOrder;
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/SchemaUtil.java b/fluss-common/src/main/java/org/apache/fluss/utils/SchemaUtil.java
index 6b66c0cd6..1815e71cd 100644
--- a/fluss-common/src/main/java/org/apache/fluss/utils/SchemaUtil.java
+++ b/fluss-common/src/main/java/org/apache/fluss/utils/SchemaUtil.java
@@ -39,6 +39,7 @@ public class SchemaUtil {
* @param expectedSchema
* @return
*/
+ // TODO
public static int[] getTargetColumns(
int[] targetColumns, Schema originSchema, Schema expectedSchema) {
if (targetColumns == null) {
@@ -57,12 +58,17 @@ public class SchemaUtil {
}
/**
- * Get the index of each value of expectedSchema from originSchema.
+ * Get the index mapping from origin schema to expected schema. This mapping can be used to
+ * convert the row with origin schema to the row with expected schema via {@code
+ * ProjectedRow.from(getIndexMapping(inputSchema, targetSchema)).replace(inputRow)}.
*
- * @param originSchema
- * @param expectedSchema
- * @return indexMapping the value of i-th means the expectedSchema's i-th column is the
- * originSchema's indexMapping[i]-th column.
+ * @param originSchema The origin schema.
+ * @param expectedSchema The expected schema.
+ * @return The index mapping array. The length of the array is the number of columns in the
+ * expected schema. Each element in the array is the index of the corresponding column in
+ * the origin schema. If a column in the expected schema does not exist in the origin
+ * schema, the corresponding element is UNEXIST_MAPPING (-1).
+ * @throws SchemaChangeException if there is a datatype mismatch between the two schemas.
*/
public static int[] getIndexMapping(Schema originSchema, Schema expectedSchema) {
List<Schema.Column> originColumns = originSchema.getColumns();
@@ -84,7 +90,7 @@ public class SchemaUtil {
if (originColumn != null
&& !Objects.equals(
expectedColumn.getDataType().copy(true),
- originColumns.get(indexMapping[i]).getDataType().copy(true))) {
+ originColumn.getDataType().copy(true))) {
throw new SchemaChangeException(
String.format(
@@ -92,7 +98,8 @@ public class SchemaUtil {
expectedColumn.getColumnId(),
expectedColumn.getName(),
expectedColumn.getDataType(),
- originColumn.getDataType()));
+ originColumn.getDataType()),
+ true);
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metadata/SchemaMetadataManager.java b/fluss-server/src/main/java/org/apache/fluss/server/metadata/SchemaMetadataManager.java
index 8b6be5e39..6bed8dfe3 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/metadata/SchemaMetadataManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/metadata/SchemaMetadataManager.java
@@ -41,6 +41,7 @@ import java.util.concurrent.ExecutionException;
* 1. latest schema of each subscribed table, updated by UpdateMetadata request. 2. history schemas
* of each table, updated by lookup from tablet server.
*/
+// TODO SchemaManager
public class SchemaMetadataManager {
private MetadataManager metadataManager;
(fluss) 07/07: WIP3
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch ci-add-column
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit ff3db43d42e64d9d0691455ad9dbccfb7f5728e5
Author: Jark Wu <ja...@apache.org>
AuthorDate: Mon Dec 1 02:35:27 2025 +0800
WIP3
---
.../fluss/flink/sink/writer/FlinkSinkWriter.java | 2 +-
.../source/reader/FlinkSourceSplitReader.java | 3 +-
.../flink/utils/FlussRowToFlinkRowConverter.java | 1 -
.../fluss/lake/iceberg/conf/HadoopUtils.java | 5 +-
.../iceberg/maintenance/IcebergRewriteITCase.java | 2 +-
.../fluss/rpc/netty/client/ServerConnection.java | 3 +-
fluss-rpc/src/main/proto/FlussApi.proto | 12 +--
.../coordinator/CompletedSnapshotStoreManager.java | 2 +-
.../coordinator/CoordinatorEventProcessor.java | 1 -
.../fluss/server/coordinator/MetadataManager.java | 8 +-
.../fluss/server/coordinator/SchemaUpdate.java | 28 +++---
.../fluss/server/coordinator/UpdateSchema.java | 28 ------
.../event/watcher/TableChangeWatcher.java | 9 --
.../server/entity/BatchRegisterLeadAndIsr.java | 2 +-
.../apache/fluss/server/kv/KvRecoverHelper.java | 99 +++++++++-------------
.../java/org/apache/fluss/server/kv/KvTablet.java | 11 +--
.../server/kv/snapshot/CompletedSnapshotStore.java | 4 +-
.../fluss/server/kv/snapshot/SnapshotLocation.java | 2 +-
...MetadataManager.java => ServerSchemaCache.java} | 7 +-
.../server/metadata/TabletServerMetadataCache.java | 15 ++--
...ManagerTest.java => ServerSchemaCacheTest.java} | 16 ++--
.../server/testutils/FlussClusterExtension.java | 6 +-
22 files changed, 98 insertions(+), 168 deletions(-)
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/FlinkSinkWriter.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/FlinkSinkWriter.java
index 97a70ce25..fe0eb28ea 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/FlinkSinkWriter.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/FlinkSinkWriter.java
@@ -110,7 +110,7 @@ public abstract class FlinkSinkWriter<InputT> implements SinkWriter<InputT> {
connection = ConnectionFactory.createConnection(flussConfig, flinkMetricRegistry);
table = connection.getTable(tablePath);
LOG.info(
- "Current Fluss Schema is {}, Table RowType is {}",
+ "Current Fluss Schema is {}, input Flink RowType is {}",
table.getTableInfo().getSchema(),
tableRowType);
sanityCheck(table.getTableInfo());
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java
index 10997e91e..a19572dc9 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java
@@ -217,8 +217,7 @@ public class FlinkSourceSplitReader implements SplitReader<RecordAndPos, SourceS
(LakeSnapshotAndFlussLogSplit) sourceSplitBase;
if (lakeSnapshotAndFlussLogSplit.isStreaming()) {
// is streaming split which has no stopping offset, we need also
- // subscribe
- // change log
+ // subscribe change log
subscribeLog(
lakeSnapshotAndFlussLogSplit,
lakeSnapshotAndFlussLogSplit.getStartingOffset());
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlussRowToFlinkRowConverter.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlussRowToFlinkRowConverter.java
index 4dc5287b0..0846b284b 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlussRowToFlinkRowConverter.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlussRowToFlinkRowConverter.java
@@ -47,7 +47,6 @@ public class FlussRowToFlinkRowConverter {
private final InternalRow.FieldGetter[] flussFieldGetters;
private final int[] indexMapping;
- // todo: remove
public FlussRowToFlinkRowConverter(RowType rowType) {
this(rowType, toFlinkRowType(rowType));
}
diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/conf/HadoopUtils.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/conf/HadoopUtils.java
index 9c86a92c1..5aad1d5bf 100644
--- a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/conf/HadoopUtils.java
+++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/conf/HadoopUtils.java
@@ -79,8 +79,7 @@ public class HadoopUtils {
}
// Approach 3: Fluss configuration
- // add all configuration key with prefix 'iceberg.hadoop.' in fluss conf to hadoop
- // conf
+ // add all configuration key with prefix 'iceberg.hadoop.' in fluss conf to hadoop conf
for (String key : flussConfiguration.keySet()) {
for (String prefix : FLUSS_CONFIG_PREFIXES) {
if (key.startsWith(prefix)) {
@@ -109,7 +108,7 @@ public class HadoopUtils {
}
/**
- * Search Hadoop configuration files in the given path, andadd them to the configuration if
+ * Search Hadoop configuration files in the given path, and add them to the configuration if
* found.
*/
private static boolean addHadoopConfIfFound(
diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/maintenance/IcebergRewriteITCase.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/maintenance/IcebergRewriteITCase.java
index 133c23322..4b6d9de71 100644
--- a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/maintenance/IcebergRewriteITCase.java
+++ b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/maintenance/IcebergRewriteITCase.java
@@ -147,7 +147,7 @@ class IcebergRewriteITCase extends FlinkIcebergTieringTestBase {
// add pos-delete and trigger compaction
rows = Arrays.asList(row(4, "v1"), row(4, "v2"));
flussRows.add(writeIcebergTableRecords(t1, t1Bucket, 6, false, rows).get(1));
- // rewritten files should fail to commit due to conflict,add check here
+ // rewritten files should fail to commit due to conflict, add check here
checkRecords(getIcebergRecords(t1), flussRows);
// 4 data file and 1 delete file
checkFileStatusInIcebergTable(t1, 4, true);
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java
index 7f4d6e84e..a09a50c09 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java
@@ -69,8 +69,7 @@ final class ServerConnection {
private final ServerNode node;
- // TODO: add max inflight requests limit like Kafka's
- // "max.in.flight.requests.per.connection"
+ // TODO: add max inflight requests limit like Kafka's "max.in.flight.requests.per.connection"
private final Map<Integer, InflightRequest> inflightRequests = MapUtils.newConcurrentHashMap();
private final CompletableFuture<Void> closeFuture = new CompletableFuture<>();
private final ConnectionMetrics connectionMetrics;
diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto
index 9feb38dc1..a08eeaba4 100644
--- a/fluss-rpc/src/main/proto/FlussApi.proto
+++ b/fluss-rpc/src/main/proto/FlussApi.proto
@@ -918,33 +918,33 @@ message PbAlterConfig {
required int32 op_type = 3; // SET=0, DELETE=1, APPEND=2, SUBTRACT=3
}
-message PbAddColumn{
+message PbAddColumn {
required string column_name = 1;
required bytes data_type_json = 2;
optional string comment = 3;
required int32 column_position_type = 4; // LAST=0,FIRST=1,AFTER=3
}
-message PbDropColumn{
+message PbDropColumn {
required string column_name = 1;
}
-message PbRenameColumn{
+message PbRenameColumn {
required string old_column_name = 1;
required string new_column_name = 2;
}
-message PbModifyColumn{
+message PbModifyColumn {
required string column_name = 1;
- required bytes data_type_json = 2;
+ optional bytes data_type_json = 2;
optional string comment = 3;
optional int32 column_position_type = 4; // LAST=0,FIRST=1,AFTER=3
}
-message PbDescribeConfig{
+message PbDescribeConfig {
required string config_key = 1;
optional string config_value = 2;
required string config_source = 3;
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java
index def57231d..b7c92289b 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java
@@ -148,7 +148,7 @@ public class CompletedSnapshotStoreManager {
() -> getAllSnapshotSize(bucket));
} else {
LOG.warn(
- "Failed toadd bucketMetricGroup for tableBucket {} when creating completed snapshot.",
+ "Failed to add bucketMetricGroup for tableBucket {} when creating completed snapshot.",
bucket);
}
return snapshotStore;
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
index fbe0986b1..d1d551121 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
@@ -556,7 +556,6 @@ public class CoordinatorEventProcessor implements EventProcessor {
private void processCreateTable(CreateTableEvent createTableEvent) {
long tableId = createTableEvent.getTableInfo().getTableId();
- int schemaId = createTableEvent.getTableInfo().getSchemaId();
// skip the table if it already exists
if (coordinatorContext.containsTableId(tableId)) {
return;
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
index 19b13aaa9..bf2cfc9fb 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
@@ -330,11 +330,7 @@ public class MetadataManager {
// validate the table column changes
if (!schemaChanges.isEmpty()) {
- UpdateSchema schemaUpdate = new SchemaUpdate(table);
- for (TableChange schemaChange : schemaChanges) {
- schemaUpdate = schemaUpdate.applySchemaChange(schemaChange);
- }
- Schema newSchema = schemaUpdate.getSchema();
+ Schema newSchema = SchemaUpdate.applySchemaChanges(table, schemaChanges);
// update the schema
zookeeperClient.registerSchema(tablePath, newSchema);
}
@@ -485,7 +481,7 @@ public class MetadataManager {
if (toEnableDataLake) {
TableInfo newTableInfo = newTableRegistration.toTableInfo(tablePath, schemaInfo);
- // if the table is lake table, we need toadd it to lake table tiering manager
+ // if the table is lake table, we need to add it to lake table tiering manager
lakeTableTieringManager.addNewLakeTable(newTableInfo);
} else if (toDisableDataLake) {
lakeTableTieringManager.removeLakeTable(newTableRegistration.tableId);
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/SchemaUpdate.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/SchemaUpdate.java
index 1cdc702d7..9efab2a51 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/SchemaUpdate.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/SchemaUpdate.java
@@ -29,12 +29,20 @@ import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
/** Schema update. */
-public class SchemaUpdate implements UpdateSchema {
+public class SchemaUpdate {
+
+ /** Apply schema changes to the given table info and return the updated schema. */
+ public static Schema applySchemaChanges(TableInfo tableInfo, List<TableChange> changes) {
+ SchemaUpdate schemaUpdate = new SchemaUpdate(tableInfo);
+ for (TableChange change : changes) {
+ schemaUpdate = schemaUpdate.applySchemaChange(change);
+ }
+ return schemaUpdate.getSchema();
+ }
+
private final List<Schema.Column> columns;
private final AtomicInteger highestFieldId;
private final List<String> primaryKeys;
- private final List<String> bucketKeys;
- private final List<String> partitionKeys;
private final Map<String, Schema.Column> existedColumns;
public SchemaUpdate(TableInfo tableInfo) {
@@ -42,15 +50,12 @@ public class SchemaUpdate implements UpdateSchema {
this.existedColumns = new HashMap<>();
this.highestFieldId = new AtomicInteger(tableInfo.getSchema().getHighestFieldId());
this.primaryKeys = tableInfo.getPrimaryKeys();
- this.bucketKeys = tableInfo.getBucketKeys();
- this.partitionKeys = tableInfo.getPartitionKeys();
this.columns.addAll(tableInfo.getSchema().getColumns());
for (Schema.Column column : columns) {
existedColumns.put(column.getName(), column);
}
}
- @Override
public Schema getSchema() {
Schema.Builder builder =
Schema.newBuilder()
@@ -63,8 +68,7 @@ public class SchemaUpdate implements UpdateSchema {
return builder.build();
}
- @Override
- public UpdateSchema applySchemaChange(TableChange columnChange) {
+ public SchemaUpdate applySchemaChange(TableChange columnChange) {
if (columnChange instanceof TableChange.AddColumn) {
return addColumn((TableChange.AddColumn) columnChange);
} else if (columnChange instanceof TableChange.ModifyColumn) {
@@ -78,7 +82,7 @@ public class SchemaUpdate implements UpdateSchema {
"Unknown column change type " + columnChange.getClass().getName());
}
- private UpdateSchema addColumn(TableChange.AddColumn addColumn) {
+ private SchemaUpdate addColumn(TableChange.AddColumn addColumn) {
if (existedColumns.containsKey(addColumn.getName())) {
throw new IllegalArgumentException(
"Column " + addColumn.getName() + " already exists.");
@@ -105,15 +109,15 @@ public class SchemaUpdate implements UpdateSchema {
return this;
}
- private UpdateSchema dropColumn(TableChange.DropColumn dropColumn) {
+ private SchemaUpdate dropColumn(TableChange.DropColumn dropColumn) {
throw new SchemaChangeException("Not support drop column now.");
}
- private UpdateSchema modifiedColumn(TableChange.ModifyColumn modifyColumn) {
+ private SchemaUpdate modifiedColumn(TableChange.ModifyColumn modifyColumn) {
throw new SchemaChangeException("Not support modify column now.");
}
- private UpdateSchema renameColumn(TableChange.RenameColumn renameColumn) {
+ private SchemaUpdate renameColumn(TableChange.RenameColumn renameColumn) {
throw new SchemaChangeException("Not support rename column now.");
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/UpdateSchema.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/UpdateSchema.java
deleted file mode 100644
index 1d51704ac..000000000
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/UpdateSchema.java
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.fluss.server.coordinator;
-
-import org.apache.fluss.metadata.Schema;
-import org.apache.fluss.metadata.TableChange;
-
-/** Schema update. */
-public interface UpdateSchema {
- Schema getSchema();
-
- UpdateSchema applySchemaChange(TableChange columnChange);
-}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/watcher/TableChangeWatcher.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/watcher/TableChangeWatcher.java
index acfb7e689..09c273e7d 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/watcher/TableChangeWatcher.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/watcher/TableChangeWatcher.java
@@ -246,16 +246,7 @@ public class TableChangeWatcher {
private void processSchemaChange(TablePath tablePath, int schemaId) {
try {
- int currentSchemaId = zooKeeperClient.getCurrentSchemaId(tablePath);
SchemaInfo schemaInfo;
- if (schemaId != currentSchemaId) {
- LOG.warn(
- "Schema id {} is not equal to current schema id {}. Skipping schema change processing.",
- schemaId,
- currentSchemaId);
- return;
- }
-
Optional<SchemaInfo> optSchema = zooKeeperClient.getSchemaById(tablePath, schemaId);
if (!optSchema.isPresent()) {
LOG.error("No schema for table {} in zookeeper.", tablePath);
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/entity/BatchRegisterLeadAndIsr.java b/fluss-server/src/main/java/org/apache/fluss/server/entity/BatchRegisterLeadAndIsr.java
index fad96a596..460562bc1 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/entity/BatchRegisterLeadAndIsr.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/entity/BatchRegisterLeadAndIsr.java
@@ -55,7 +55,7 @@ public class BatchRegisterLeadAndIsr {
tableBucket, leaderAndIsr, partitionName, liveReplicas));
} else {
throw new IllegalArgumentException(
- "Try toadd a bucket with different tableId or partitionId in collection when try to batch register to Zookeeper."
+ "Try to add a bucket with different tableId or partitionId in collection when try to batch register to Zookeeper."
+ "batch tableId="
+ tableId
+ " batch partitionId:"
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java
index 1f1c4192d..96bb0eba8 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java
@@ -66,15 +66,6 @@ public class KvRecoverHelper {
private InternalRow.FieldGetter[] currentFieldGetters;
- public KvRecoverHelper(
- KvTablet kvTablet,
- LogTablet logTablet,
- long recoverPointOffset,
- KvRecoverContext recoverContext,
- KvFormat kvFormat) {
- throw new UnsupportedOperationException();
- }
-
public KvRecoverHelper(
KvTablet kvTablet,
LogTablet logTablet,
@@ -136,64 +127,52 @@ public class KvRecoverHelper {
FetchIsolation fetchIsolation,
ThrowingConsumer<KeyValueAndLogOffset, Exception> resumeRecordConsumer)
throws Exception {
- long nextFetchOffset = startFetchOffset;
- while (true) {
- LogRecords logRecords =
- logTablet
- .read(
- nextFetchOffset,
- recoverContext.maxFetchLogSizeInRecoverKv,
- fetchIsolation,
- true,
- null)
- .getRecords();
- if (logRecords == MemoryLogRecords.EMPTY) {
- break;
- }
-
- for (LogRecordBatch logRecordBatch : logRecords.batches()) {
- // short schemaId = logRecordBatch.schemaId();
- // if (currentSchemaId == null) {
- // initSchema(schemaId);
- // } else if (currentSchemaId != schemaId) {
- // throw new KvStorageException(
- // String.format(
- // "Can't recover kv tablet for table bucket from
- // log %s since the schema changes from schema id %d to schema id %d. "
- // + "Currently, schema change is not
- // supported.",
- // recoverContext.tableBucket, currentSchemaId,
- // schemaId));
- // }
+ try (LogRecordReadContext readContext =
+ LogRecordReadContext.createArrowReadContext(
+ currentRowType, currentSchemaId, schemaGetter)) {
+ long nextFetchOffset = startFetchOffset;
+ while (true) {
+ LogRecords logRecords =
+ logTablet
+ .read(
+ nextFetchOffset,
+ recoverContext.maxFetchLogSizeInRecoverKv,
+ fetchIsolation,
+ true,
+ null)
+ .getRecords();
+ if (logRecords == MemoryLogRecords.EMPTY) {
+ break;
+ }
- // todo: currentRowType和currentSchemaId无法对齐
- try (LogRecordReadContext readContext =
- LogRecordReadContext.createArrowReadContext(
- currentRowType, currentSchemaId, schemaGetter);
- CloseableIterator<LogRecord> logRecordIter =
- logRecordBatch.records(readContext)) {
- while (logRecordIter.hasNext()) {
- LogRecord logRecord = logRecordIter.next();
- if (logRecord.getChangeType() != ChangeType.UPDATE_BEFORE) {
- InternalRow logRow = logRecord.getRow();
- byte[] key = keyEncoder.encodeKey(logRow);
- byte[] value = null;
- if (logRecord.getChangeType() != ChangeType.DELETE) {
- // the log row format may not compatible with kv row format,
- // e.g, arrow vs. compacted, thus needs a conversion here.
- BinaryRow row = toKvRow(logRecord.getRow());
- // todo: short value是否会有问题，感觉可以先check一下
- value = ValueEncoder.encodeValue(currentSchemaId.shortValue(), row);
+ for (LogRecordBatch logRecordBatch : logRecords.batches()) {
+ try (CloseableIterator<LogRecord> logRecordIter =
+ logRecordBatch.records(readContext)) {
+ while (logRecordIter.hasNext()) {
+ LogRecord logRecord = logRecordIter.next();
+ if (logRecord.getChangeType() != ChangeType.UPDATE_BEFORE) {
+ InternalRow logRow = logRecord.getRow();
+ byte[] key = keyEncoder.encodeKey(logRow);
+ byte[] value = null;
+ if (logRecord.getChangeType() != ChangeType.DELETE) {
+ // the log row format may not compatible with kv row format,
+ // e.g, arrow vs. compacted, thus needs a conversion here.
+ BinaryRow row = toKvRow(logRecord.getRow());
+ value =
+ ValueEncoder.encodeValue(
+ currentSchemaId.shortValue(), row);
+ }
+ resumeRecordConsumer.accept(
+ new KeyValueAndLogOffset(
+ key, value, logRecord.logOffset()));
}
- resumeRecordConsumer.accept(
- new KeyValueAndLogOffset(key, value, logRecord.logOffset()));
}
}
+ nextFetchOffset = logRecordBatch.nextLogOffset();
}
- nextFetchOffset = logRecordBatch.nextLogOffset();
}
+ return nextFetchOffset;
}
- return nextFetchOffset;
}
// TODO: this is very in-efficient, because the conversion is CPU heavy. Should be optimized in
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java
index 176e749f5..70099878e 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java
@@ -83,7 +83,6 @@ import java.util.concurrent.Executor;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
-import static org.apache.fluss.utils.SchemaUtil.getTargetColumns;
import static org.apache.fluss.utils.concurrent.LockUtils.inReadLock;
import static org.apache.fluss.utils.concurrent.LockUtils.inWriteLock;
@@ -271,7 +270,7 @@ public final class KvTablet {
// schema1: old data's schema -> valueDecoder
// schema2: new data's schema
// schema3: current schema (also the latest schema)
- // schema1 <= schema2 <= current schema id
+ // (schema1 <= or >= schema2) <= current schema id
short latestSchemaId = (short) schemaInfo.getSchemaId();
short schemaIdOfNewData = kvRecords.schemaId();
if (schemaIdOfNewData > latestSchemaId || schemaIdOfNewData < 0) {
@@ -282,14 +281,10 @@ public final class KvTablet {
+ latestSchemaId);
}
Schema schemaOfNewData = schemaGetter.getSchema(schemaIdOfNewData);
+ // we only support ADD COLUMN, so targetColumns is fine to be used directly
RowMerger currentMerger =
rowMerger.configureTargetColumns(
- schemaIdOfNewData == latestSchemaId
- ? targetColumns
- : getTargetColumns(
- targetColumns, schemaOfNewData, latestSchema),
- latestSchemaId,
- latestSchema);
+ targetColumns, latestSchemaId, latestSchema);
RowType currentRowType = latestSchema.getRowType();
WalBuilder walBuilder = createWalBuilder(latestSchemaId, currentRowType);
walBuilder.setWriterState(kvRecords.writerId(), kvRecords.batchSequence());
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStore.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStore.java
index b8d941221..3e93cc46a 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStore.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStore.java
@@ -99,7 +99,7 @@ public class CompletedSnapshotStore {
* Synchronously writes the new snapshots to snapshot handle store and asynchronously removes
* older ones.
*
- * @param snapshot Completed snapshot toadd.
+ * @param snapshot Completed snapshot to add.
*/
@VisibleForTesting
CompletedSnapshot addSnapshotAndSubsumeOldestOne(
@@ -116,7 +116,7 @@ public class CompletedSnapshotStore {
completedSnapshotHandleStore.add(
snapshot.getTableBucket(), snapshot.getSnapshotID(), completedSnapshotHandle);
- // Nowadd the new one. If it fails, we don't want to lose existing data.
+ // Now add the new one. If it fails, we don't want to lose existing data.
completedSnapshots.addLast(snapshot);
// Remove completed snapshot from queue and snapshotStateHandleStore, not discard.
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotLocation.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotLocation.java
index e09c93b22..fe517f268 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotLocation.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotLocation.java
@@ -328,7 +328,7 @@ public class SnapshotLocation {
Exception latestException = null;
for (int attempt = 0; attempt < 10; attempt++) {
try {
- // todo: mayadd entropy injection?
+ // todo: may add entropy injection?
this.kvFilePath = createFilePath();
this.outStream = fs.create(kvFilePath, FileSystem.WriteMode.NO_OVERWRITE);
return;
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metadata/SchemaMetadataManager.java b/fluss-server/src/main/java/org/apache/fluss/server/metadata/ServerSchemaCache.java
similarity index 97%
rename from fluss-server/src/main/java/org/apache/fluss/server/metadata/SchemaMetadataManager.java
rename to fluss-server/src/main/java/org/apache/fluss/server/metadata/ServerSchemaCache.java
index 6bed8dfe3..0f98b092d 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/metadata/SchemaMetadataManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/metadata/ServerSchemaCache.java
@@ -41,15 +41,14 @@ import java.util.concurrent.ExecutionException;
* 1. latest schema of each subscribed table, updated by UpdateMetadata request. 2. history schemas
* of each table, updated by lookup from tablet server.
*/
-// TODO SchemaManager
-public class SchemaMetadataManager {
+public class ServerSchemaCache {
- private MetadataManager metadataManager;
+ private final MetadataManager metadataManager;
private final Map<TablePath, SchemaInfo> latestSchemaByTablePath;
private final Map<TablePath, Integer> subscriberCounters;
private final Cache<TableSchemaKey, Schema> schemaCache;
- public SchemaMetadataManager(MetadataManager metadataManager) {
+ public ServerSchemaCache(MetadataManager metadataManager) {
this.metadataManager = metadataManager;
// thread safe is guaranteed by subscriberCounters.
this.subscriberCounters = MapUtils.newConcurrentHashMap();
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java b/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java
index 6a11a53ae..9a73975ce 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java
@@ -68,13 +68,13 @@ public class TabletServerMetadataCache implements ServerMetadataCache {
private final MetadataManager metadataManager;
- private final SchemaMetadataManager schemaMetadataManager;
+ private final ServerSchemaCache serverSchemaCache;
// todo: replace this in test with schemaMetadataManager.
public TabletServerMetadataCache(MetadataManager metadataManager) {
this.serverMetadataSnapshot = ServerMetadataSnapshot.empty();
this.metadataManager = metadataManager;
- this.schemaMetadataManager = new SchemaMetadataManager(metadataManager);
+ this.serverSchemaCache = new ServerSchemaCache(metadataManager);
}
@Override
@@ -142,12 +142,12 @@ public class TabletServerMetadataCache implements ServerMetadataCache {
public SchemaGetter subscribeWithInitialSchema(
TablePath tablePath, int initialSchemaId, Schema initialSchema) {
- return schemaMetadataManager.subscribeWithInitialSchema(
+ return serverSchemaCache.subscribeWithInitialSchema(
tablePath, initialSchemaId, initialSchema);
}
public void updateLatestSchema(TablePath tablePath, SchemaInfo schemaInfo) {
- schemaMetadataManager.updateLatestSchema(
+ serverSchemaCache.updateLatestSchema(
tablePath, (short) schemaInfo.getSchemaId(), schemaInfo.getSchema());
}
@@ -207,8 +207,7 @@ public class TabletServerMetadataCache implements ServerMetadataCache {
// todo: apply schema id and schema info if needs
int schemaId = tableInfo.getSchemaId();
Schema schema = tableInfo.getSchema();
- schemaMetadataManager.updateLatestSchema(
- tablePath, (short) schemaId, schema);
+ serverSchemaCache.updateLatestSchema(tablePath, (short) schemaId, schema);
if (tableId == DELETED_TABLE_ID) {
Long removedTableId = tableIdByPath.remove(tablePath);
@@ -437,7 +436,7 @@ public class TabletServerMetadataCache implements ServerMetadataCache {
}
@VisibleForTesting
- public SchemaMetadataManager getSchemaMetadataManager() {
- return schemaMetadataManager;
+ public ServerSchemaCache getSchemaMetadataManager() {
+ return serverSchemaCache;
}
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/metadata/SchemaMetadataManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/metadata/ServerSchemaCacheTest.java
similarity index 94%
rename from fluss-server/src/test/java/org/apache/fluss/server/metadata/SchemaMetadataManagerTest.java
rename to fluss-server/src/test/java/org/apache/fluss/server/metadata/ServerSchemaCacheTest.java
index 672910b04..f72277ba8 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/metadata/SchemaMetadataManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/metadata/ServerSchemaCacheTest.java
@@ -44,13 +44,13 @@ import static org.apache.fluss.record.TestData.DATA2_TABLE_DESCRIPTOR;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-/** Test for {@link SchemaMetadataManager}. */
-public class SchemaMetadataManagerTest {
+/** Test for {@link ServerSchemaCache}. */
+public class ServerSchemaCacheTest {
@Test
void testPublishAndsubscribeSchemaChange() {
- SchemaMetadataManager manager =
- new SchemaMetadataManager(new TestingMetadataManager(Collections.emptyList()));
+ ServerSchemaCache manager =
+ new ServerSchemaCache(new TestingMetadataManager(Collections.emptyList()));
SchemaGetter subscriber1 =
manager.subscribeWithInitialSchema(
@@ -94,8 +94,8 @@ public class SchemaMetadataManagerTest {
DATA2_TABLE_DESCRIPTOR,
System.currentTimeMillis(),
System.currentTimeMillis());
- SchemaMetadataManager manager =
- new SchemaMetadataManager(
+ ServerSchemaCache manager =
+ new ServerSchemaCache(
new TestingMetadataManager(Arrays.asList(DATA1_TABLE_INFO, tableInfo2)));
SchemaGetter schemaGetter =
manager.subscribeWithInitialSchema(DATA1_TABLE_PATH, (short) 2, DATA2_SCHEMA);
@@ -107,8 +107,8 @@ public class SchemaMetadataManagerTest {
@Test
void testUnsubscribeSchemaChange() {
- SchemaMetadataManager manager =
- new SchemaMetadataManager(new TestingMetadataManager(Collections.emptyList()));
+ ServerSchemaCache manager =
+ new ServerSchemaCache(new TestingMetadataManager(Collections.emptyList()));
SchemaGetter subscriber1 =
manager.subscribeWithInitialSchema(
new TablePath("test_tb", "test_table_1"), (short) 1, DATA1_SCHEMA);
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java b/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java
index 37f3dd6cb..ef1cb92ba 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java
@@ -51,8 +51,8 @@ import org.apache.fluss.server.coordinator.MetadataManager;
import org.apache.fluss.server.entity.NotifyLeaderAndIsrData;
import org.apache.fluss.server.kv.snapshot.CompletedSnapshot;
import org.apache.fluss.server.kv.snapshot.CompletedSnapshotHandle;
-import org.apache.fluss.server.metadata.SchemaMetadataManager;
import org.apache.fluss.server.metadata.ServerInfo;
+import org.apache.fluss.server.metadata.ServerSchemaCache;
import org.apache.fluss.server.metadata.TabletServerMetadataCache;
import org.apache.fluss.server.replica.Replica;
import org.apache.fluss.server.replica.ReplicaManager;
@@ -678,10 +678,10 @@ public final class FlussClusterExtension
assertThat(leaderAndIsrOpt).isPresent();
int leader = leaderAndIsrOpt.get().leader();
TabletServer tabletServer = getTabletServerById(leader);
- SchemaMetadataManager schemaMetadataManager =
+ ServerSchemaCache serverSchemaCache =
tabletServer.getMetadataCache().getSchemaMetadataManager();
Map<TablePath, SchemaInfo> latestSchemaByTablePath =
- schemaMetadataManager.getLatestSchemaByTablePath();
+ serverSchemaCache.getLatestSchemaByTablePath();
assertThat(latestSchemaByTablePath).containsKey(tablePath);
assertThat(latestSchemaByTablePath.get(tablePath).getSchemaId())
.isEqualTo(schemaId);
(fluss) 02/07: WIP1
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch ci-add-column
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 6295cf7d94fae0864c1b9cc1100dda4401fc6997
Author: Jark Wu <ja...@apache.org>
AuthorDate: Sun Nov 30 21:20:15 2025 +0800
WIP1
---
.../apache/fluss/record/DefaultLogRecordBatch.java | 96 ++++++----------------
1 file changed, 27 insertions(+), 69 deletions(-)
diff --git a/fluss-common/src/main/java/org/apache/fluss/record/DefaultLogRecordBatch.java b/fluss-common/src/main/java/org/apache/fluss/record/DefaultLogRecordBatch.java
index b6ed40eeb..8bf27e10e 100644
--- a/fluss-common/src/main/java/org/apache/fluss/record/DefaultLogRecordBatch.java
+++ b/fluss-common/src/main/java/org/apache/fluss/record/DefaultLogRecordBatch.java
@@ -21,7 +21,6 @@ import org.apache.fluss.annotation.PublicEvolving;
import org.apache.fluss.exception.CorruptMessageException;
import org.apache.fluss.memory.MemorySegment;
import org.apache.fluss.metadata.LogFormat;
-import org.apache.fluss.row.InternalRow;
import org.apache.fluss.row.ProjectedRow;
import org.apache.fluss.row.arrow.ArrowReader;
import org.apache.fluss.row.columnar.ColumnarRow;
@@ -34,13 +33,13 @@ import org.apache.fluss.utils.CloseableIterator;
import org.apache.fluss.utils.MurmurHashUtils;
import org.apache.fluss.utils.crc.Crc32C;
+import javax.annotation.Nullable;
+
import java.nio.ByteBuffer;
import java.util.NoSuchElementException;
-import java.util.function.Function;
import static org.apache.fluss.record.LogRecordBatchFormat.BASE_OFFSET_OFFSET;
import static org.apache.fluss.record.LogRecordBatchFormat.COMMIT_TIMESTAMP_OFFSET;
-import static org.apache.fluss.record.LogRecordBatchFormat.LENGTH_LENGTH;
import static org.apache.fluss.record.LogRecordBatchFormat.LENGTH_OFFSET;
import static org.apache.fluss.record.LogRecordBatchFormat.LOG_MAGIC_VALUE_V1;
import static org.apache.fluss.record.LogRecordBatchFormat.LOG_OVERHEAD;
@@ -213,7 +212,6 @@ public class DefaultLogRecordBatch implements LogRecordBatch {
}
@Override
- @SuppressWarnings("unchecked")
public CloseableIterator<LogRecord> records(ReadContext context) {
if (getRecordCount() == 0) {
return CloseableIterator.emptyIterator();
@@ -223,26 +221,18 @@ public class DefaultLogRecordBatch implements LogRecordBatch {
long timestamp = commitTimestamp();
LogFormat logFormat = context.getLogFormat();
RowType rowType = context.getRowType(schemaId);
- Function<InternalRow, InternalRow> outputRowProcessor =
- (row) -> {
- ProjectedRow projectedRow = context.getOutputProjectedRow(schemaId);
- if (projectedRow != null) {
- return projectedRow.replaceRow(row);
- } else {
- return row;
- }
- };
switch (logFormat) {
case ARROW:
return columnRecordIterator(
rowType,
- outputRowProcessor,
+ context.getOutputProjectedRow(schemaId),
context.getVectorSchemaRoot(schemaId),
context.getBufferAllocator(),
timestamp);
case INDEXED:
- return rowRecordIterator(rowType, outputRowProcessor, timestamp);
+ return rowRecordIterator(
+ rowType, context.getOutputProjectedRow(schemaId), timestamp);
default:
throw new IllegalArgumentException("Unsupported log format: " + logFormat);
}
@@ -270,9 +260,7 @@ public class DefaultLogRecordBatch implements LogRecordBatch {
}
private CloseableIterator<LogRecord> rowRecordIterator(
- RowType rowType,
- Function<InternalRow, InternalRow> outputRowProcessor,
- long timestamp) {
+ RowType rowType, @Nullable ProjectedRow outputProjection, long timestamp) {
DataType[] fieldTypes = rowType.getChildren().toArray(new DataType[0]);
return new LogRecordIterator() {
int position = DefaultLogRecordBatch.this.position + recordBatchHeaderSize(magic);
@@ -281,16 +269,20 @@ public class DefaultLogRecordBatch implements LogRecordBatch {
@Override
protected LogRecord readNext(long baseOffset) {
IndexedLogRecord logRecord =
- ProjectedIndexedLogRecord.readFrom(
- segment,
- position,
- baseOffset + rowId,
- timestamp,
- fieldTypes,
- outputRowProcessor);
+ IndexedLogRecord.readFrom(
+ segment, position, baseOffset + rowId, timestamp, fieldTypes);
rowId++;
position += logRecord.getSizeInBytes();
- return logRecord;
+ if (outputProjection == null) {
+ return logRecord;
+ } else {
+ // apply projection
+ return new GenericRecord(
+ logRecord.logOffset(),
+ logRecord.timestamp(),
+ logRecord.getChangeType(),
+ outputProjection.replaceRow(logRecord.getRow()));
+ }
}
@Override
@@ -305,7 +297,7 @@ public class DefaultLogRecordBatch implements LogRecordBatch {
private CloseableIterator<LogRecord> columnRecordIterator(
RowType rowType,
- Function<InternalRow, InternalRow> outputRowProcessor,
+ @Nullable ProjectedRow outputProjection,
VectorSchemaRoot root,
BufferAllocator allocator,
long timestamp) {
@@ -319,7 +311,7 @@ public class DefaultLogRecordBatch implements LogRecordBatch {
ArrowReader reader =
ArrowUtils.createArrowReader(
segment, arrowOffset, arrowLength, root, allocator, rowType);
- return new ArrowLogRecordIterator(reader, timestamp, outputRowProcessor) {
+ return new ArrowLogRecordIterator(reader, timestamp, outputProjection) {
@Override
protected ChangeType getChangeType(int rowId) {
return ChangeType.APPEND_ONLY;
@@ -337,7 +329,7 @@ public class DefaultLogRecordBatch implements LogRecordBatch {
ArrowReader reader =
ArrowUtils.createArrowReader(
segment, arrowOffset, arrowLength, root, allocator, rowType);
- return new ArrowLogRecordIterator(reader, timestamp, outputRowProcessor) {
+ return new ArrowLogRecordIterator(reader, timestamp, outputProjection) {
@Override
protected ChangeType getChangeType(int rowId) {
return changeTypeVector.getChangeType(rowId);
@@ -351,15 +343,13 @@ public class DefaultLogRecordBatch implements LogRecordBatch {
private final ArrowReader reader;
private final long timestamp;
private int rowId = 0;
- private final Function<InternalRow, InternalRow> outputRowProcessor;
+ @Nullable private final ProjectedRow outputProjection;
private ArrowLogRecordIterator(
- ArrowReader reader,
- long timestamp,
- Function<InternalRow, InternalRow> outputRowProcessor) {
+ ArrowReader reader, long timestamp, @Nullable ProjectedRow outputProjection) {
this.reader = reader;
this.timestamp = timestamp;
- this.outputRowProcessor = outputRowProcessor;
+ this.outputProjection = outputProjection;
}
protected abstract ChangeType getChangeType(int rowId);
@@ -377,7 +367,9 @@ public class DefaultLogRecordBatch implements LogRecordBatch {
baseOffset + rowId,
timestamp,
getChangeType(rowId),
- outputRowProcessor.apply(originalRow));
+ outputProjection == null
+ ? originalRow
+ : outputProjection.replaceRow(originalRow));
rowId++;
return record;
}
@@ -447,38 +439,4 @@ public class DefaultLogRecordBatch implements LogRecordBatch {
throw new UnsupportedOperationException();
}
}
-
- private static class ProjectedIndexedLogRecord extends IndexedLogRecord {
- private final Function<InternalRow, InternalRow> outputRowProcessor;
-
- ProjectedIndexedLogRecord(
- long logOffset,
- long timestamp,
- DataType[] fieldTypes,
- Function<InternalRow, InternalRow> outputRowProcessor) {
- super(logOffset, timestamp, fieldTypes);
- this.outputRowProcessor = outputRowProcessor;
- }
-
- public static IndexedLogRecord readFrom(
- MemorySegment segment,
- int position,
- long logOffset,
- long logTimestamp,
- DataType[] colTypes,
- Function<InternalRow, InternalRow> outputRowProcessor) {
- int sizeInBytes = segment.getInt(position);
- ProjectedIndexedLogRecord logRecord =
- new ProjectedIndexedLogRecord(
- logOffset, logTimestamp, colTypes, outputRowProcessor);
- logRecord.pointTo(segment, position, sizeInBytes + LENGTH_LENGTH);
- return logRecord;
- }
-
- @Override
- public InternalRow getRow() {
- InternalRow row = super.getRow();
- return outputRowProcessor.apply(row);
- }
- }
}