Skip to content

Commit 5a9d3d1

Browse files
LiebingYuLiebingYu
committed
[WIP] Rescale bucket
1 parent b6dc1a3 commit 5a9d3d1

File tree

15 files changed

+895
-21
lines changed

15 files changed

+895
-21
lines changed

‎fluss-client/src/test/java/com/alibaba/fluss/client/admin/ClientToServerITCaseBase.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,9 @@ public abstract class ClientToServerITCaseBase {
8383
@BeforeEach
8484
protected void setup() throws Exception {
8585
clientConf = FLUSS_CLUSTER_EXTENSION.getClientConfig();
86+
clientConf.set(
87+
ConfigOptions.CLIENT_WRITER_BUCKET_NO_KEY_ASSIGNER,
88+
ConfigOptions.NoKeyAssigner.ROUND_ROBIN);
8689
conn = ConnectionFactory.createConnection(clientConf);
8790
admin = conn.getAdmin();
8891
}
@@ -235,6 +238,13 @@ public static void waitAllReplicasReady(long tableId, int expectBucketCount) {
235238
}
236239
}
237240

241+
public static void waitAllReplicasReady(long tableId, long partitionId, int expectBucketCount) {
242+
for (int i = 0; i < expectBucketCount; i++) {
243+
FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(
244+
new TableBucket(tableId, partitionId, i));
245+
}
246+
}
247+
238248
protected static void verifyRows(
239249
RowType rowType,
240250
Map<Long, List<InternalRow>> actualRows,

‎fluss-client/src/test/java/com/alibaba/fluss/client/admin/FlussAdminITCase.java‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,7 @@
6060
import com.alibaba.fluss.metadata.TablePath;
6161
import com.alibaba.fluss.server.kv.snapshot.CompletedSnapshot;
6262
import com.alibaba.fluss.server.kv.snapshot.KvSnapshotHandle;
63+
import com.alibaba.fluss.server.zk.data.TableAssignment;
6364
import com.alibaba.fluss.types.DataTypes;
6465

6566
import org.junit.jupiter.api.BeforeEach;
@@ -197,6 +198,54 @@ void testGetTableInfoAndSchema() throws Exception {
197198
.isBetween(timestampBeforeCreate, timestampAfterCreate);
198199
}
199200

201+
@Test
202+
void testAlterTableBucket() throws Exception {
203+
// create table
204+
TablePath tablePath = TablePath.of("test_db", "alter_table_bucket");
205+
admin.createTable(tablePath, DEFAULT_TABLE_DESCRIPTOR, false).get();
206+
207+
TableInfo tableInfo = admin.getTableInfo(tablePath).get();
208+
209+
TableAssignment tableAssignment =
210+
FLUSS_CLUSTER_EXTENSION
211+
.getZooKeeperClient()
212+
.getTableAssignment(tableInfo.getTableId())
213+
.get();
214+
System.out.println(tableAssignment);
215+
216+
TableDescriptor existingTableDescriptor = tableInfo.toTableDescriptor();
217+
218+
TableDescriptor newTableDescriptor =
219+
TableDescriptor.builder()
220+
.schema(existingTableDescriptor.getSchema())
221+
.comment(existingTableDescriptor.getComment().orElse("test table"))
222+
.partitionedBy(existingTableDescriptor.getPartitionKeys())
223+
.distributedBy(
224+
existingTableDescriptor
225+
.getTableDistribution()
226+
.get()
227+
.getBucketCount()
228+
.get()
229+
+ 1,
230+
existingTableDescriptor.getBucketKeys())
231+
.properties(existingTableDescriptor.getProperties())
232+
.customProperties(existingTableDescriptor.getCustomProperties())
233+
.build();
234+
// alter table
235+
admin.alterTable(tablePath, newTableDescriptor, false).get();
236+
237+
TableInfo alteredTableInfo = admin.getTableInfo(tablePath).get();
238+
TableDescriptor alteredTableDescriptor = alteredTableInfo.toTableDescriptor();
239+
// assertThat(alteredTableDescriptor).isEqualTo(newTableDescriptor);
240+
241+
TableAssignment tableAssignment1 =
242+
FLUSS_CLUSTER_EXTENSION
243+
.getZooKeeperClient()
244+
.getTableAssignment(tableInfo.getTableId())
245+
.get();
246+
System.out.println(tableAssignment1);
247+
}
248+
200249
@Test
201250
void testAlterTable() throws Exception {
202251
// create table

‎fluss-client/src/test/java/com/alibaba/fluss/client/table/FlussTableITCase.java‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -645,6 +645,61 @@ void testPutAndPoll(String kvFormat) throws Exception {
645645
verifyAppendOrPut(false, "ARROW", kvFormat);
646646
}
647647

648+
@Test
649+
void testAlterTableBucket() throws Exception {
650+
TableDescriptor DATA1_TABLE_DESCRIPTOR =
651+
TableDescriptor.builder().schema(DATA1_SCHEMA).distributedBy(1).build();
652+
createTable(DATA1_TABLE_PATH, DATA1_TABLE_DESCRIPTOR, false);
653+
654+
long tableId;
655+
try (Table table = conn.getTable(DATA1_TABLE_PATH)) {
656+
tableId = table.getTableInfo().getTableId();
657+
658+
AppendWriter appendWriter = table.newAppend().createWriter();
659+
660+
for (int i = 0; i < 10; i++) {
661+
appendWriter.append(row(i, "a")).get();
662+
}
663+
664+
try (LogScanner logScanner = createLogScanner(table)) {
665+
subscribeFromBeginning(logScanner, table);
666+
667+
ScanRecords scanRecords = logScanner.poll(Duration.ofSeconds(1));
668+
for (ScanRecord record : scanRecords) {
669+
System.out.println(record);
670+
}
671+
}
672+
}
673+
674+
System.out.println("==");
675+
676+
TableDescriptor DATA2_TABLE_DESCRIPTOR =
677+
TableDescriptor.builder().schema(DATA1_SCHEMA).distributedBy(2).build();
678+
admin.alterTable(DATA1_TABLE_PATH, DATA2_TABLE_DESCRIPTOR, false);
679+
680+
waitAllReplicasReady(tableId, 2);
681+
682+
conn = ConnectionFactory.createConnection(clientConf);
683+
try (Table table = conn.getTable(DATA1_TABLE_PATH)) {
684+
AppendWriter appendWriter = table.newAppend().createWriter();
685+
686+
for (int i = 0; i < 10; i++) {
687+
appendWriter.append(row(i, "a")).get();
688+
}
689+
690+
try (LogScanner logScanner = createLogScanner(table)) {
691+
subscribeFromBeginning(logScanner, table);
692+
693+
ScanRecords scanRecords = logScanner.poll(Duration.ofSeconds(1));
694+
for (TableBucket tableBucket : scanRecords.buckets()) {
695+
for (ScanRecord record : scanRecords.records(tableBucket)) {
696+
System.out.println("Bucket: " + tableBucket + " record: " + record);
697+
}
698+
}
699+
}
700+
}
701+
}
702+
648703
void verifyAppendOrPut(boolean append, String logFormat, @Nullable String kvFormat)
649704
throws Exception {
650705
Schema schema =

‎fluss-client/src/test/java/com/alibaba/fluss/client/table/PartitionedTableITCase.java‎

Lines changed: 99 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,8 +17,12 @@
1717

1818
package com.alibaba.fluss.client.table;
1919

20+
import com.alibaba.fluss.client.ConnectionFactory;
2021
import com.alibaba.fluss.client.admin.ClientToServerITCaseBase;
2122
import com.alibaba.fluss.client.lookup.Lookuper;
23+
import com.alibaba.fluss.client.table.scanner.ScanRecord;
24+
import com.alibaba.fluss.client.table.scanner.log.LogScanner;
25+
import com.alibaba.fluss.client.table.scanner.log.ScanRecords;
2226
import com.alibaba.fluss.client.table.writer.AppendWriter;
2327
import com.alibaba.fluss.client.table.writer.UpsertWriter;
2428
import com.alibaba.fluss.config.ConfigOptions;
@@ -27,14 +31,18 @@
2731
import com.alibaba.fluss.metadata.PartitionInfo;
2832
import com.alibaba.fluss.metadata.PhysicalTablePath;
2933
import com.alibaba.fluss.metadata.Schema;
34+
import com.alibaba.fluss.metadata.TableBucket;
3035
import com.alibaba.fluss.metadata.TableDescriptor;
36+
import com.alibaba.fluss.metadata.TableInfo;
3137
import com.alibaba.fluss.metadata.TablePath;
3238
import com.alibaba.fluss.row.GenericRow;
3339
import com.alibaba.fluss.row.InternalRow;
3440
import com.alibaba.fluss.types.DataTypes;
3541

3642
import org.junit.jupiter.api.Test;
3743

44+
import javax.annotation.Nullable;
45+
3846
import java.time.Duration;
3947
import java.util.ArrayList;
4048
import java.util.HashMap;
@@ -142,6 +150,86 @@ void testPartitionedLogTable() throws Exception {
142150
verifyPartitionLogs(table, schema.getRowType(), expectPartitionAppendRows);
143151
}
144152

153+
@Test
154+
void testAlterPartitionTableBucket() throws Exception {
155+
TablePath tablePath = TablePath.of("test_db_1", "test_static_partitioned_log_table_1");
156+
Schema schema = createPartitionedTable(tablePath, false, 1);
157+
TableInfo tableInfo = admin.getTableInfo(tablePath).get();
158+
159+
List<PartitionInfo> partitionInfos = admin.listPartitionInfos(tablePath).get();
160+
assertThat(partitionInfos.isEmpty()).isTrue();
161+
162+
// add three partitions.
163+
for (int i = 0; i < 1; i++) {
164+
admin.createPartition(tablePath, newPartitionSpec("c", "c" + i), false).get();
165+
}
166+
partitionInfos = admin.listPartitionInfos(tablePath).get();
167+
assertThat(partitionInfos.size()).isEqualTo(1);
168+
169+
Table table = conn.getTable(tablePath);
170+
AppendWriter appendWriter = table.newAppend().createWriter();
171+
int recordsPerPartition = 5;
172+
Map<Long, List<InternalRow>> expectPartitionAppendRows = new HashMap<>();
173+
for (PartitionInfo partitionInfo : partitionInfos) {
174+
String partitionName = partitionInfo.getPartitionName();
175+
long partitionId = partitionInfo.getPartitionId();
176+
for (int j = 0; j < recordsPerPartition; j++) {
177+
InternalRow row = row(j, "a" + j, partitionName);
178+
appendWriter.append(row);
179+
expectPartitionAppendRows
180+
.computeIfAbsent(partitionId, k -> new ArrayList<>())
181+
.add(row);
182+
}
183+
}
184+
appendWriter.flush();
185+
186+
// then, let's verify the logs
187+
verifyPartitionLogs(table, schema.getRowType(), expectPartitionAppendRows);
188+
189+
TableDescriptor partitionTableDescriptor =
190+
TableDescriptor.builder()
191+
.schema(schema)
192+
.distributedBy(2)
193+
.partitionedBy("c")
194+
.build();
195+
admin.alterTable(tablePath, partitionTableDescriptor, false);
196+
197+
for (PartitionInfo partitionInfo : partitionInfos) {
198+
waitAllReplicasReady(tableInfo.getTableId(), partitionInfo.getPartitionId(), 2);
199+
}
200+
201+
conn = ConnectionFactory.createConnection(clientConf);
202+
table = conn.getTable(tablePath);
203+
appendWriter = table.newAppend().createWriter();
204+
for (PartitionInfo partitionInfo : partitionInfos) {
205+
String partitionName = partitionInfo.getPartitionName();
206+
long partitionId = partitionInfo.getPartitionId();
207+
for (int j = 0; j < recordsPerPartition; j++) {
208+
InternalRow row = row(j, "a" + j, partitionName);
209+
appendWriter.append(row);
210+
expectPartitionAppendRows
211+
.computeIfAbsent(partitionId, k -> new ArrayList<>())
212+
.add(row);
213+
}
214+
}
215+
appendWriter.flush();
216+
217+
try (LogScanner logScanner = table.newScan().createLogScanner()) {
218+
for (PartitionInfo partitionInfo : partitionInfos) {
219+
logScanner.subscribeFromBeginning(partitionInfo.getPartitionId(), 0);
220+
logScanner.subscribeFromBeginning(partitionInfo.getPartitionId(), 1);
221+
}
222+
223+
ScanRecords scanRecords = logScanner.poll(Duration.ofSeconds(1));
224+
for (TableBucket scanBucket : scanRecords.buckets()) {
225+
List<ScanRecord> records = scanRecords.records(scanBucket);
226+
for (ScanRecord scanRecord : records) {
227+
System.out.println("Bucket: " + scanBucket + ", Record: " + scanRecord);
228+
}
229+
}
230+
}
231+
}
232+
145233
@Test
146234
void testWriteToNonExistsPartitionWhenDisabledDynamicPartition() throws Exception {
147235
clientConf.set(ConfigOptions.CLIENT_WRITER_DYNAMIC_CREATE_PARTITION_ENABLED, false);
@@ -234,6 +322,12 @@ void testCreatePartitionExceedMaxPartitionNumber() throws Exception {
234322

235323
private Schema createPartitionedTable(TablePath tablePath, boolean isPrimaryTable)
236324
throws Exception {
325+
return createPartitionedTable(tablePath, isPrimaryTable, null);
326+
}
327+
328+
private Schema createPartitionedTable(
329+
TablePath tablePath, boolean isPrimaryTable, @Nullable Integer bucketCount)
330+
throws Exception {
237331
Schema.Builder schemaBuilder =
238332
Schema.newBuilder()
239333
.column("a", DataTypes.INT())
@@ -250,7 +344,11 @@ private Schema createPartitionedTable(TablePath tablePath, boolean isPrimaryTabl
250344
Schema schema = schemaBuilder.build();
251345

252346
TableDescriptor partitionTableDescriptor =
253-
TableDescriptor.builder().schema(schema).partitionedBy("c").build();
347+
TableDescriptor.builder()
348+
.schema(schema)
349+
.distributedBy(bucketCount)
350+
.partitionedBy("c")
351+
.build();
254352
createTable(tablePath, partitionTableDescriptor, false);
255353
return schema;
256354
}

0 commit comments

Comments
 (0)