update

Author: yinshuangfei

pkg/sink/cloudstorage/table_definition.go

@@ -337,7 +337,16 @@ func (t *TableDefinition) GenerateSchemaFilePath(useTableIDAsPath bool, tableID
 		return "", err
 	}
 	if !t.IsTableSchema() && t.Table != "" {
-		log.Panic("invalid table definition", zap.Any("tableDef", t))
+		return "", errors.ErrInternalCheckFailed.GenWithStackByArgs(
+			"invalid table definition",
+		)
+	}
+	if useTableIDAsPath && t.Table != "" {
+		if tableID <= 0 {
+			return "", errors.ErrInternalCheckFailed.GenWithStackByArgs(
+				"invalid table id for table-id path",
+			)
+		}
 	}
 	table := generateTablePath(t.Table, tableID, useTableIDAsPath)
 	return generateSchemaFilePath(t.Schema, table, t.TableVersion, checksum), nil

pkg/sink/cloudstorage/table_definition_test.go

@@ -502,6 +502,33 @@ func TestTableDefinitionGenFilePath(t *testing.T) {
 	require.Equal(t, "schema1/12345/meta/schema_100_3752767265.json", tablePath)
 }
 
+func TestGenerateSchemaFilePathValidation(t *testing.T) {
+	t.Parallel()
+
+	def, _ := generateTableDef()
+
+	// nil receiver - would panic before entering, skip as Go handles this
+	// empty schema
+	emptySchemaDef := &TableDefinition{Schema: "", Table: "t1", TableVersion: 100, TotalColumns: 1, Columns: []TableCol{{}}}
+	_, err := emptySchemaDef.GenerateSchemaFilePath(false, 0)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "schema cannot be empty")
+
+	// zero table version
+	zeroVersionDef := &TableDefinition{Schema: "s1", Table: "t1", TableVersion: 0, TotalColumns: 1, Columns: []TableCol{{}}}
+	_, err = zeroVersionDef.GenerateSchemaFilePath(false, 0)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "table version cannot be zero")
+
+	// use-table-id-as-path with invalid tableID
+	_, err = def.GenerateSchemaFilePath(true, 0)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "tableID must be positive")
+	_, err = def.GenerateSchemaFilePath(true, -1)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "tableID must be positive")
+}
+
 func TestTableDefinitionSum32(t *testing.T) {
 	t.Parallel()