Add Enhanced Ingestion Mode to genddl Tool

cmd/genddl/config_enhanced_ingestion.json.example

@@ -0,0 +1,18 @@
+{
+  "mode": "enhanced_ingestion",
+  "scale_factor": "1000",
+  "file_format": "parquet",
+  "compression_method": "uncompressed",
+  "workload": "tpcds",
+  "workload_definition": "tpc-ds",
+  "iceberg": true,
+  "source_file_format": "CSV",
+  "source_catalog": "source_catalog_name",
+  "target_catalog": "target_catalog_name",
+  "source_schema": "source_schema_name",
+  "target_schema": "target_schema_name",
+  "s3_source_location": "s3a://bucket-name/path/to/source/data",
+  "s3_target_location": "s3a://bucket-name/path/to/target/data",
+  "engine": "presto",
+  "session_variables": {}
+}

cmd/genddl/create_source_table.sql.tmpl

@@ -0,0 +1,45 @@
+{{- /*gotype:pbench/cmd/genddl.Schema*/ -}}
+{{- /* This template is used for ENHANCED_INGESTION mode - creates source tables */ -}}
+{{- if .SourceSchema }}
+{{- if .SourceCatalog }}
+CREATE SCHEMA IF NOT EXISTS {{ .SourceCatalog }}.{{ .SourceSchema }}
+WITH (
+  location = '{{ .S3SourceLocation }}/'
+);
+
+USE {{ .SourceCatalog }}.{{ .SourceSchema }};
+{{- else }}
+CREATE SCHEMA IF NOT EXISTS hive.{{ .SourceSchema }}
+WITH (
+  location = '{{ .S3SourceLocation }}/'
+);
+
+USE hive.{{ .SourceSchema }};
+{{- end }}
+
+{{ range .Tables }}
+CREATE TABLE IF NOT EXISTS {{ .Name }} (
+{{- $first := true }}
+{{- range .Columns }}
+  {{- if $first }}
+  {{- $first = false }}
+  {{- else }},
+  {{- end }}
+  {{- if eq $.SourceFileFormat "CSV" }}
+  {{ .Name }} varchar
+  {{- else }}
+  {{ .Name }} {{ .Type }}
+  {{- end }}
+{{- end }})
+WITH (
+  external_location = '{{ $.S3SourceLocation }}/{{ .Name }}/',
+  {{- if eq $.SourceFileFormat "CSV" }}
+  format = 'CSV',
+  csv_separator = '|'
+  {{- else }}
+  format = 'TEXTFILE',
+  textfile_field_delim = '|'
+  {{- end }}
+);
+{{ end }}
+{{- end }}

cmd/genddl/create_target_table.sql.tmpl

@@ -0,0 +1,61 @@
+{{- /*gotype:pbench/cmd/genddl.Schema*/ -}}
+{{- /* This template is used for ENHANCED_INGESTION mode - creates target tables */ -}}
+{{- if .TargetSchema }}
+{{- if .TargetCatalog }}
+CREATE SCHEMA IF NOT EXISTS {{ .TargetCatalog }}.{{ .TargetSchema }}
+WITH (
+  location = '{{ .S3TargetLocation }}/'
+);
+
+USE {{ .TargetCatalog }}.{{ .TargetSchema }};
+{{- else }}
+{{- if .Iceberg }}
+CREATE SCHEMA IF NOT EXISTS iceberg.{{ .TargetSchema }}
+WITH (
+  location = '{{ .S3TargetLocation }}/'
+);
+
+USE iceberg.{{ .TargetSchema }};
+{{- else }}
+CREATE SCHEMA IF NOT EXISTS hive.{{ .TargetSchema }}
+WITH (
+  location = '{{ .S3TargetLocation }}/'
+);
+
+USE hive.{{ .TargetSchema }};
+{{- end }}
+{{- end }}
+
+{{ range .Tables }}
+CREATE TABLE IF NOT EXISTS {{ .Name }} (
+{{- $first := true }}
+{{- range .Columns }}
+    {{- if $first }}
+        {{- $first = false }}
+    {{- else -}}
+        ,
+    {{- end }}
+    {{ .Name }} {{ .Type }}
+{{- end }}
+)
+{{- if eq $.Engine "spark" }}
+USING iceberg
+{{- if .Partitioned }}
+PARTITIONED BY ({{ .LastColumn.Name }})
+{{- end }}
+TBLPROPERTIES ('write.target-file-size-bytes' = 268435456, 'write.parquet.row-group-size-bytes' = 67108864)
+LOCATION '{{ $.S3TargetLocation }}/{{ .Name }}';
+{{- else }}
+WITH (
+    format = 'PARQUET'
+    {{- if $.Partitioned }}
+    {{- if $.Iceberg}}
+    , partitioning = array['{{ .LastColumn.Name }}']
+    {{- else if .Partitioned }}
+    , partitioned_by = array['{{ .LastColumn.Name }}']
+    {{- end }}
+    {{- end }}
+);
+{{- end }}
+{{ end }}
+{{- end }}

cmd/genddl/insert_table.sql.tmpl

@@ -3,6 +3,48 @@
 SET SESSION {{ $key }}='{{ $value }}';
 {{ end }}
 
+{{- if .TargetSchema }}
+{{- /* ENHANCED_INGESTION MODE */ -}}
+{{- if .TargetCatalog }}
+USE {{ .TargetCatalog }}.{{ .TargetSchema }};
+{{- else }}
+{{- if .Iceberg }}
+USE iceberg.{{ .TargetSchema }};
+{{- else }}
+USE hive.{{ .TargetSchema }};
+{{- end }}
+{{- end }}
+
+{{ range .InsertTables -}}
+{{- if $.TargetCatalog }}
+INSERT INTO {{ $.TargetCatalog }}.{{ $.TargetSchema }}.{{ .Name }}
+{{- else }}
+INSERT INTO {{ .Name }}
+{{- end }}
+{{- if eq $.SourceFileFormat "TEXTFILE" }}
+SELECT * FROM {{ if $.SourceCatalog }}{{ $.SourceCatalog }}.{{ $.SourceSchema }}.{{ .Name }}{{ else }}hive.{{ $.SourceSchema }}.{{ .Name }}{{ end }};
+{{- else }}
+SELECT
+{{- $first := true }}
+{{- range .Columns }}
+  {{- if $first }}
+  {{- $first = false }}
+  {{- else }},
+  {{- end }}
+  {{- if eq $.SourceFileFormat "CSV" }}
+    {{- if .IsVarchar }}
+  NULLIF({{ .Name }}, '') AS {{ .Name }}
+    {{- else }}
+  CAST(NULLIF({{ .Name }}, '') AS {{ .Type }}) AS {{ .Name }}
+    {{- end }}
+  {{- end }}
+{{- end }}
+FROM {{ if $.SourceCatalog }}{{ $.SourceCatalog }}.{{ $.SourceSchema }}.{{ .Name }}{{ else }}hive.{{ $.SourceSchema }}.{{ .Name }}{{ end }};
+{{- end }}
+
+{{ end }}
+{{- else }}
+{{- /* LEGACY MODE (old workflow) */ -}}
 {{- if .Iceberg }}
 USE iceberg.{{ .SchemaName }};
 {{- else }}
@@ -31,6 +73,8 @@ SELECT * FROM {{ if $.Iceberg }}iceberg{{ else }}hive{{ end }}.{{ $.Uncompressed
 {{- end }}
 
 {{ end }}
+{{- end }}
+
 {{- range .InsertTables -}}
 ANALYZE {{ .Name }};
 {{ end -}}