badalprasadsingh commented on code in PR #1244:
URL: https://github.com/apache/iceberg-go/pull/1244#discussion_r3530932652
##########
catalog/sql/sql.go:
##########
@@ -247,12 +286,141 @@ func (c *Catalog) DropSQLTables(ctx context.Context)
error {
}
func (c *Catalog) ensureTablesExist() error {
- return c.CreateSQLTables(context.Background())
+ ctx := context.Background()
+ if err := c.CreateSQLTables(ctx); err != nil {
+ return err
+ }
+
+ return c.migrateV0Schema(ctx)
+}
+
+func (c *Catalog) detectSchemaVersion(ctx context.Context) error {
+ hasCol, err := c.icebergTypeColumnExists(ctx)
+ if err != nil {
+ return fmt.Errorf("detecting catalog schema version: %w", err)
+ }
+
+ if hasCol {
+ c.schemaVersion = schemaV1
+ } else {
+ c.schemaVersion = schemaV0
+ }
+
+ return nil
+}
+
+func (c *Catalog) migrateV0Schema(ctx context.Context) error {
+ hasCol, err := c.icebergTypeColumnExists(ctx)
+ if err != nil {
+ return fmt.Errorf("V0 schema migration: column probe failed:
%w", err)
+ }
+ if hasCol {
+ c.schemaVersion = schemaV1
+
+ return nil
+ }
+
+ if !strings.EqualFold(c.props.Get(SchemaVersionKey, ""),
SchemaVersionV1) {
+ log.Printf("WARNING: iceberg_tables is on the legacy V0 schema
(no iceberg_type column); "+
+ "set %q=%q to migrate to V1. Operating in V0-compatible
mode (column-free queries).",
+ SchemaVersionKey, SchemaVersionV1)
+ c.schemaVersion = schemaV0
+
+ return nil
+ }
+
+ ddl, err := addIcebergTypeColumnDDL(c.db.Dialect().Name())
+ if err != nil {
+ return err
+ }
+
+ if _, err := c.db.ExecContext(ctx, ddl); err != nil {
+ has, reprobeErr := c.icebergTypeColumnExists(ctx)
+ if has {
+ c.schemaVersion = schemaV1
+
+ return nil
+ }
+ if reprobeErr != nil {
+ return fmt.Errorf("migration of V0 schema failed: %w
(re-probe also failed: %v)", err, reprobeErr)
+ }
+
+ return fmt.Errorf("migration of V0 schema failed: %w", err)
+ }
+
+ c.schemaVersion = schemaV1
+
+ return nil
+}
+
+func addIcebergTypeColumnDDL(name dialect.Name) (string, error) {
+ switch name {
+ case dialect.PG, dialect.SQLite:
+ return "ALTER TABLE iceberg_tables ADD COLUMN iceberg_type
VARCHAR", nil
+ case dialect.MySQL:
+ return "ALTER TABLE iceberg_tables ADD COLUMN iceberg_type
VARCHAR(255)", nil
+ case dialect.MSSQL:
+ return "ALTER TABLE iceberg_tables ADD iceberg_type
VARCHAR(255)", nil
+ case dialect.Oracle:
+ return "ALTER TABLE iceberg_tables ADD (iceberg_type
VARCHAR2(255))", nil
+ default:
+ return "", fmt.Errorf("unsupported dialect for V0 migration:
%s", name)
+ }
+}
+
+func icebergTypeColumnExistsQuery(name dialect.Name) (string, error) {
+ switch name {
+ case dialect.PG:
+ return `SELECT 1 FROM information_schema.columns
+ WHERE table_schema = current_schema()
+ AND table_name = 'iceberg_tables'
+ AND column_name = 'iceberg_type' LIMIT 1`, nil
+ case dialect.MySQL:
+ return `SELECT 1 FROM information_schema.columns
+ WHERE table_schema = DATABASE()
+ AND table_name = 'iceberg_tables'
+ AND column_name = 'iceberg_type' LIMIT 1`, nil
+ case dialect.MSSQL:
+ return `SELECT TOP 1 1 FROM information_schema.columns
+ WHERE table_schema = SCHEMA_NAME()
+ AND table_name = 'iceberg_tables'
+ AND column_name = 'iceberg_type'`, nil
+ case dialect.SQLite:
+ return `SELECT 1 FROM pragma_table_info('iceberg_tables')
+ WHERE name = 'iceberg_type' LIMIT 1`, nil
+ case dialect.Oracle:
+ return `SELECT 1 FROM user_tab_columns
+ WHERE UPPER(table_name) = 'ICEBERG_TABLES'
+ AND UPPER(column_name) = 'ICEBERG_TYPE'`, nil
+ default:
+ return "", fmt.Errorf("unsupported dialect for V0 migration:
%s", name)
+ }
+}
+
+func (c *Catalog) icebergTypeColumnExists(ctx context.Context) (bool, error) {
+ query, err := icebergTypeColumnExistsQuery(c.db.Dialect().Name())
+ if err != nil {
+ return false, err
+ }
+
+ var exists int
+ if err := c.db.QueryRowContext(ctx, query).Scan(&exists); err != nil {
Review Comment:
Done.
##########
catalog/sql/sql.go:
##########
@@ -1013,6 +1225,10 @@ func (c *Catalog) DropView(ctx context.Context,
identifier table.Identifier) err
// CheckViewExists returns true if a view exists in the catalog.
func (c *Catalog) CheckViewExists(ctx context.Context, identifier
table.Identifier) (bool, error) {
+ if c.isV0() {
+ return false, errViewsUnsupportedOnV0
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
