Skip to content

Commit b9e77b6

Browse files
authored
feat(fuse): support partitioned table layout (#20143)
* feat(fuse): organize data by partition key * feat(fuse): prune segments by partition values * fix(fuse): preserve partitioned update statistics * test(fuse): stabilize partition pruning assertions * test(fuse): cover partition-aware recluster * fix(query): emit replayable partitioned table DDL * fix(ast): preserve Iceberg partition syntax * fix(query): validate partition table options * perf(fuse): scan partition boundaries by column type * test(fuse): document fragmented partition writes * test(fuse): avoid topology-dependent segment count * fix(fuse): prune day partitions from timestamp ranges * ci(sqllogic): update partition cluster expectation * fix(fuse): flush final partition segment * ci(sqllogic): avoid topology-dependent block count * test(fuse): add missing coverage for PARTITION BY - sqllogictest: verify SET OPTIONS partition_by rejection, MODIFY COLUMN partition-key rejection, unrelated-filter no-pruning, DELETE+compact boundary preservation, and CTAS partition boundary on initial write - unit: partition_values/same_partition fallback cases (None stats, mismatched key id, min!=max prefix) - unit: PartitionPruner::should_keep returns true conservatively when segment has no cluster statistics * ci(sqllogic): isolate unrelated partition filter
1 parent a5fe8de commit b9e77b6

63 files changed

Lines changed: 2217 additions & 225 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

Cargo.lock

Lines changed: 1 addition & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

src/query/ast/src/ast/statements/table.rs

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -162,7 +162,7 @@ pub struct CreateTableStmt {
162162
pub uri_location: Option<UriLocation>,
163163
pub cluster_by: Option<ClusterOption>,
164164
pub table_options: BTreeMap<String, String>,
165-
pub iceberg_table_partition: Option<Vec<Identifier>>,
165+
pub partition_by: Option<Vec<Expr>>,
166166
pub table_properties: Option<BTreeMap<String, String>>,
167167
pub as_query: Option<Box<Query>>,
168168
pub table_type: TableType,
@@ -211,6 +211,12 @@ impl Display for CreateTableStmt {
211211
write!(f, " {uri_location}")?;
212212
}
213213

214+
if let Some(partition_by) = &self.partition_by {
215+
write!(f, " PARTITION BY(")?;
216+
write_comma_separated_list(f, partition_by)?;
217+
write!(f, ")")?;
218+
}
219+
214220
if let Some(cluster_by) = &self.cluster_by {
215221
write!(f, " {cluster_by}")?;
216222
}
@@ -221,12 +227,6 @@ impl Display for CreateTableStmt {
221227
write_space_separated_string_map(f, &self.table_options)?;
222228
}
223229

224-
if let Some(iceberg_table_partition) = &self.iceberg_table_partition {
225-
write!(f, " PARTITION BY(")?;
226-
write_comma_separated_list(f, iceberg_table_partition)?;
227-
write!(f, ")")?;
228-
}
229-
230230
if let Some(table_properties) = &self.table_properties {
231231
write!(f, " PROPERTIES(")?;
232232
write_space_separated_string_map(f, table_properties)?;

src/query/ast/src/parser/statement.rs

Lines changed: 14 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1131,16 +1131,22 @@ pub fn statement_body(i: Input) -> IResult<Statement> {
11311131
})
11321132
},
11331133
);
1134+
let create_table_partition_by = map(
1135+
rule! {
1136+
#table_option ~ PARTITION ~ ^BY ~ ^"(" ~ ^#comma_separated_list1(expr) ~ ^")"
1137+
},
1138+
|(table_options, _, _, _, exprs, _)| (table_options, exprs),
1139+
);
11341140
let create_table = map_res(
11351141
rule! {
11361142
CREATE ~ ( OR ~ ^REPLACE )? ~ (TEMP| TEMPORARY|TRANSIENT)? ~ TABLE ~ ( IF ~ ^NOT ~ ^EXISTS )?
11371143
~ #dot_separated_idents_1_to_3
11381144
~ #create_table_source?
11391145
~ ( #engine )?
11401146
~ ( #uri_location )?
1147+
~ #create_table_partition_by?
11411148
~ ( CLUSTER ~ ^BY ~ LINEAR? ~ ^"(" ~ ^#comma_separated_list1(expr) ~ ^")" )?
11421149
~ ( #table_option )?
1143-
~ ( PARTITION ~ ^BY ~ ^"(" ~ ^#comma_separated_list1(ident) ~ ^")" )?
11441150
~ ( PROPERTIES ~ #connection_options )?
11451151
~ ( AS ~ ^#query )?
11461152
},
@@ -1154,9 +1160,9 @@ pub fn statement_body(i: Input) -> IResult<Statement> {
11541160
source,
11551161
engine,
11561162
uri_location,
1163+
opt_partition_by,
11571164
opt_cluster_by,
11581165
opt_table_options,
1159-
opt_iceberg_table_partition_by,
11601166
opt_table_properties,
11611167
opt_as_query,
11621168
)| {
@@ -1168,6 +1174,10 @@ pub fn statement_body(i: Input) -> IResult<Statement> {
11681174
Some(TEMP) | Some(TEMPORARY) => TableType::Temporary,
11691175
_ => unreachable!(),
11701176
};
1177+
let (mut table_options, partition_by) = opt_partition_by
1178+
.map(|(options, exprs)| (options, Some(exprs)))
1179+
.unwrap_or_default();
1180+
table_options.extend(opt_table_options.unwrap_or_default());
11711181
Ok(Statement::CreateTable(CreateTableStmt {
11721182
create_option,
11731183
catalog,
@@ -1179,9 +1189,8 @@ pub fn statement_body(i: Input) -> IResult<Statement> {
11791189
cluster_by: opt_cluster_by.map(|(_, _, _, _, exprs, _)| ClusterOption {
11801190
cluster_exprs: exprs,
11811191
}),
1182-
table_options: opt_table_options.unwrap_or_default(),
1183-
iceberg_table_partition: opt_iceberg_table_partition_by
1184-
.map(|(_, _, _, cols, _)| cols),
1192+
table_options,
1193+
partition_by,
11851194
table_properties: opt_table_properties.map(|(_, properties)| properties),
11861195
as_query: opt_as_query.map(|(_, query)| Box::new(query)),
11871196
table_type,

src/query/ast/src/visit/statement_table.rs

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -149,9 +149,9 @@ impl Walk for CreateTableStmt {
149149
if let Some(cluster_by) = &self.cluster_by {
150150
try_walk!(cluster_by.walk(visitor));
151151
}
152-
if let Some(partitions) = &self.iceberg_table_partition {
153-
for ident in partitions {
154-
try_walk!(ident.walk(visitor));
152+
if let Some(partitions) = &self.partition_by {
153+
for expr in partitions {
154+
try_walk!(expr.walk(visitor));
155155
}
156156
}
157157
if let Some(as_query) = &self.as_query {
@@ -173,9 +173,9 @@ impl WalkMut for CreateTableStmt {
173173
if let Some(cluster_by) = &mut self.cluster_by {
174174
try_walk!(cluster_by.walk_mut(visitor));
175175
}
176-
if let Some(partitions) = &mut self.iceberg_table_partition {
177-
for ident in partitions {
178-
try_walk!(ident.walk_mut(visitor));
176+
if let Some(partitions) = &mut self.partition_by {
177+
for expr in partitions {
178+
try_walk!(expr.walk_mut(visitor));
179179
}
180180
}
181181
if let Some(as_query) = &mut self.as_query {

src/query/ast/tests/it/parser.rs

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1393,6 +1393,32 @@ fn test_file_format_trim_space_option() {
13931393
assert!(displayed.contains("TRIM_SPACE = true".to_uppercase().as_str()));
13941394
}
13951395

1396+
#[test]
1397+
fn test_create_table_options_before_partition_by() {
1398+
let cases = [
1399+
"CREATE TABLE t(c INT) ENGINE=ICEBERG LOCATION='s3://bucket/path' CONNECTION_NAME='conn' PARTITION BY (c)",
1400+
"CREATE TABLE iceberg.db.t(c INT) LOCATION='s3://bucket/path' PARTITION BY (c)",
1401+
"CREATE TABLE t(a INT) ENGINE=FUSE ROW_PER_BLOCK=1 PARTITION BY (a)",
1402+
"CREATE TABLE t(a INT) ROW_PER_BLOCK=1 PARTITION BY (a)",
1403+
];
1404+
for sql in cases {
1405+
let tokens = tokenize_sql(sql).unwrap();
1406+
let (stmt, _) = parse_sql(&tokens, Dialect::PostgreSQL).unwrap();
1407+
1408+
let displayed = stmt.to_string();
1409+
let displayed_uppercase = displayed.to_uppercase();
1410+
let partition_pos = displayed_uppercase.find("PARTITION BY").unwrap();
1411+
for option in ["LOCATION", "CONNECTION_NAME", "ROW_PER_BLOCK"] {
1412+
if let Some(option_pos) = displayed_uppercase.find(option) {
1413+
assert!(partition_pos < option_pos);
1414+
}
1415+
}
1416+
1417+
let tokens = tokenize_sql(&displayed).unwrap();
1418+
parse_sql(&tokens, Dialect::PostgreSQL).unwrap();
1419+
}
1420+
}
1421+
13961422
#[test]
13971423
fn test_stage_local_filesystem_uri_errors() {
13981424
let cases = [

src/query/ast/tests/it/testdata/stmt-error.txt

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -94,8 +94,10 @@ error:
9494
--> SQL:1:55
9595
|
9696
1 | create table a (c1 decimal(38), c2 int) partition by ();
97-
| ------ ^ unexpected `)`, expecting <Ident>, <LiteralString>, or `IDENTIFIER`
98-
| |
97+
| ------ ^
98+
| | |
99+
| | expecting `<LiteralString>`, '<LiteralCodeString>', '<LiteralInteger>', '<LiteralFloat>', 'TRUE', 'FALSE', or more ...
100+
| | while parsing expression
99101
| while parsing `CREATE [OR REPLACE] TABLE [IF NOT EXISTS] [<database>.]<table> [<source>] [<table_options>]`
100102

101103

0 commit comments

Comments
 (0)