Skip to content

Commit 23ccc3b

Browse files
mjschleichclaude
mjschleich
and
claude
committed
Add Apache Iceberg export support to LQP
Add ExportIcebergConfig to the Export oneof in transactions.proto with fields: catalog_uri, namespace, table_name, catalog_properties (warehouse, token, credential), schema, prefix, target_file_size_bytes, and compression. Grammar rules, parsers, printers, and tests are included for all three SDKs (Python, Go, Julia). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1 parent 95f3c18 commit 23ccc3b

17 files changed

Lines changed: 14037 additions & 12451 deletions

File tree

‎meta/src/meta/grammar.y‎

Lines changed: 100 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -116,6 +116,8 @@
116116
%nonterm export_csv_config transactions.ExportCSVConfig
117117
%nonterm export_csv_path String
118118
%nonterm export_csv_source transactions.ExportCSVSource
119+
%nonterm export_iceberg_catalog_properties transactions.IcebergCatalogProperties
120+
%nonterm export_iceberg_config transactions.ExportIcebergConfig
119121
%nonterm false logic.Disjunction
120122
%nonterm ffi logic.FFI
121123
%nonterm ffi_args Sequence[logic.Abstraction]
@@ -199,6 +201,7 @@
199201
%validator_ignore_completeness BeTreeLocator
200202
%validator_ignore_completeness BeTreeConfig
201203
%validator_ignore_completeness ExportCSVColumns
204+
%validator_ignore_completeness IcebergCatalogProperties
202205

203206
%%
204207

@@ -1202,7 +1205,12 @@ abort
12021205
export
12031206
: "(" "export" export_csv_config ")"
12041207
construct: $$ = transactions.Export(csv_config=$3)
1205-
deconstruct: $3: transactions.ExportCSVConfig = $$.csv_config
1208+
deconstruct if builtin.has_proto_field($$, 'csv_config'):
1209+
$3: transactions.ExportCSVConfig = $$.csv_config
1210+
| "(" "export_iceberg" export_iceberg_config ")"
1211+
construct: $$ = transactions.Export(iceberg_config=$3)
1212+
deconstruct if builtin.has_proto_field($$, 'iceberg_config'):
1213+
$3: transactions.ExportIcebergConfig = $$.iceberg_config
12061214

12071215
export_csv_config
12081216
: "(" "export_csv_config_v2" export_csv_path export_csv_source csv_config ")"
@@ -1241,6 +1249,24 @@ export_csv_source
12411249
deconstruct if builtin.has_proto_field($$, 'table_def'):
12421250
$3: logic.RelationId = $$.table_def
12431251

1252+
export_iceberg_config
1253+
: "(" "export_iceberg_config" "(" "catalog_uri" STRING ")" "(" "namespace" STRING* ")" "(" "table_name" STRING ")" export_iceberg_catalog_properties "(" "schema" STRING ")" config_dict? ")"
1254+
construct: $$ = construct_export_iceberg_config_from_optional($5, $9, $13, $15, $18, $20)
1255+
deconstruct:
1256+
$5: String = $$.catalog_uri
1257+
$9: Sequence[String] = $$.namespace
1258+
$13: String = $$.table_name
1259+
$15: transactions.IcebergCatalogProperties = $$.catalog_properties
1260+
$18: String = $$.schema
1261+
$20: Optional[Sequence[Tuple[String, logic.Value]]] = deconstruct_export_iceberg_config_optional($$)
1262+
1263+
export_iceberg_catalog_properties
1264+
: "(" "catalog_properties" "(" "warehouse" STRING ")" config_dict? ")"
1265+
construct: $$ = construct_iceberg_catalog_properties_from_optional($5, $7)
1266+
deconstruct:
1267+
$5: String = $$.warehouse
1268+
$7: Optional[Sequence[Tuple[String, logic.Value]]] = deconstruct_iceberg_catalog_properties_optional($$)
1269+
12441270

12451271
%%
12461272

@@ -1558,6 +1584,79 @@ def deconstruct_export_csv_config(msg: transactions.ExportCSVConfig) -> List[Tup
15581584
return builtin.list_sort(result)
15591585

15601586

1587+
def construct_export_iceberg_config_from_optional(
1588+
catalog_uri: String,
1589+
namespace: Sequence[String],
1590+
table_name: String,
1591+
catalog_properties: transactions.IcebergCatalogProperties,
1592+
schema: String,
1593+
config_dict: Optional[Sequence[Tuple[String, logic.Value]]],
1594+
) -> transactions.ExportIcebergConfig:
1595+
prefix: String = ""
1596+
target_file_size_bytes: int = _extract_value_int64(None, 0)
1597+
compression: String = ""
1598+
if config_dict is not None:
1599+
config: Dict[String, logic.Value] = builtin.dict_from_list(builtin.unwrap_option(config_dict))
1600+
prefix = _extract_value_string(builtin.dict_get(config, "prefix"), "")
1601+
target_file_size_bytes = _extract_value_int64(builtin.dict_get(config, "target_file_size_bytes"), 0)
1602+
compression = _extract_value_string(builtin.dict_get(config, "compression"), "")
1603+
return transactions.ExportIcebergConfig(
1604+
catalog_uri=catalog_uri,
1605+
namespace=namespace,
1606+
table_name=table_name,
1607+
catalog_properties=catalog_properties,
1608+
schema=schema,
1609+
prefix=builtin.some(prefix),
1610+
target_file_size_bytes=builtin.some(target_file_size_bytes),
1611+
compression=compression,
1612+
)
1613+
1614+
1615+
def deconstruct_export_iceberg_config_optional(
1616+
msg: transactions.ExportIcebergConfig,
1617+
) -> Optional[Sequence[Tuple[String, logic.Value]]]:
1618+
result: List[Tuple[String, logic.Value]] = list[Tuple[String, logic.Value]]()
1619+
if builtin.unwrap_option(msg.prefix) != "":
1620+
builtin.list_push(result, builtin.tuple("prefix", _make_value_string(builtin.unwrap_option(msg.prefix))))
1621+
if builtin.unwrap_option(msg.target_file_size_bytes) != 0:
1622+
builtin.list_push(result, builtin.tuple("target_file_size_bytes", _make_value_int64(builtin.unwrap_option(msg.target_file_size_bytes))))
1623+
if msg.compression != "":
1624+
builtin.list_push(result, builtin.tuple("compression", _make_value_string(msg.compression)))
1625+
if builtin.length(result) == 0:
1626+
return None
1627+
return builtin.some(builtin.list_sort(result))
1628+
1629+
1630+
def construct_iceberg_catalog_properties_from_optional(
1631+
warehouse: String,
1632+
config_dict: Optional[Sequence[Tuple[String, logic.Value]]],
1633+
) -> transactions.IcebergCatalogProperties:
1634+
token: String = ""
1635+
credential: String = ""
1636+
if config_dict is not None:
1637+
config: Dict[String, logic.Value] = builtin.dict_from_list(builtin.unwrap_option(config_dict))
1638+
token = _extract_value_string(builtin.dict_get(config, "token"), "")
1639+
credential = _extract_value_string(builtin.dict_get(config, "credential"), "")
1640+
return transactions.IcebergCatalogProperties(
1641+
warehouse=warehouse,
1642+
token=builtin.some(token),
1643+
credential=builtin.some(credential),
1644+
)
1645+
1646+
1647+
def deconstruct_iceberg_catalog_properties_optional(
1648+
msg: transactions.IcebergCatalogProperties,
1649+
) -> Optional[Sequence[Tuple[String, logic.Value]]]:
1650+
result: List[Tuple[String, logic.Value]] = list[Tuple[String, logic.Value]]()
1651+
if builtin.unwrap_option(msg.token) != "":
1652+
builtin.list_push(result, builtin.tuple("token", _make_value_string(builtin.unwrap_option(msg.token))))
1653+
if builtin.unwrap_option(msg.credential) != "":
1654+
builtin.list_push(result, builtin.tuple("credential", _make_value_string(builtin.unwrap_option(msg.credential))))
1655+
if builtin.length(result) == 0:
1656+
return None
1657+
return builtin.some(builtin.list_sort(result))
1658+
1659+
15611660
def deconstruct_relation_id_string(msg: logic.RelationId) -> String:
15621661
name: Optional[String] = builtin.relation_id_to_string(msg)
15631662
return builtin.unwrap_option(name)

‎proto/relationalai/lqp/v1/transactions.proto‎

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -116,6 +116,25 @@ message ExportCSVSource {
116116
}
117117
}
118118

119+
// Iceberg Export config
120+
121+
message IcebergCatalogProperties {
122+
string warehouse = 1;
123+
optional string token = 2;
124+
optional string credential = 3;
125+
}
126+
127+
message ExportIcebergConfig {
128+
string catalog_uri = 1;
129+
repeated string namespace = 2;
130+
string table_name = 3;
131+
IcebergCatalogProperties catalog_properties = 4;
132+
string schema = 5;
133+
optional string prefix = 6; // Default: "data"
134+
optional int64 target_file_size_bytes = 7; // Default: 512 MB
135+
string compression = 8; // UNCOMPRESSED, SNAPPY, GZIP, LZ4, ZSTD
136+
}
137+
119138
//
120139
// Read operations
121140
//
@@ -142,8 +161,7 @@ message Output {
142161
message Export {
143162
oneof export_config {
144163
ExportCSVConfig csv_config = 1;
145-
146-
// TODO: support JSON export
164+
ExportIcebergConfig iceberg_config = 2;
147165
}
148166
}
149167

0 commit comments

Comments
 (0)