Merge remote-tracking branch 'upstream/develop' into develop

Author: chriswessels

CHANGELOG.md

@@ -5,6 +5,10 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## v4.6.7
+
+* Previous fix for ordering issue was not properly handling blocks batched together and was working at block level. Fix now extended to work when also inside a batch.
+
 ## v4.6.6
 
 ### `substreams-sink-sql run`

db_changes/db/db.go

@@ -47,6 +47,8 @@ type Loader struct {
 
 	testTx *TestTx // used for testing: if non-nil, 'loader.BeginTx()' will return this object instead of a real *sql.Tx
 	dsn    *DSN
+
+	batchOrdinal uint64 // Counter for ordinals within the current batch, resets on flush
 }
 
 func NewLoader(
@@ -467,6 +469,13 @@ func (l *Loader) GetDSN() *DSN {
 	return l.dsn
 }
 
+// NextBatchOrdinal returns the next ordinal for the current batch and increments the counter
+func (l *Loader) NextBatchOrdinal() uint64 {
+	ordinal := l.batchOrdinal
+	l.batchOrdinal++
+	return ordinal
+}
+
 type obfuscatedString string
 
 func (s obfuscatedString) String() string {

db_changes/db/flush.go

@@ -82,4 +82,5 @@ func (l *Loader) reset() {
 	for entriesPair := l.entries.Oldest(); entriesPair != nil; entriesPair = entriesPair.Next() {
 		l.entries.Set(entriesPair.Key, NewOrderedMap[string, *Operation]())
 	}
+	l.batchOrdinal = 0
 }

db_changes/db/operations.go

@@ -92,16 +92,11 @@ func (o *Operation) mergeData(newData map[string]string) error {
 }
 
 // mergeOperation merges another operation into this one, keeping the lowest ordinal
-func (o *Operation) mergeOperation(otherOrdinal uint64, otherData map[string]string) error {
+func (o *Operation) mergeOperation(otherData map[string]string) error {
 	if o.opType == OperationTypeDelete {
 		return fmt.Errorf("unable to merge operation for a delete operation")
 	}
 
-	// Keep the lowest ordinal
-	if otherOrdinal < o.ordinal {
-		o.ordinal = otherOrdinal
-	}
-
 	return o.mergeData(otherData)
 }
 

db_changes/db/ops.go

@@ -11,7 +11,7 @@ import (
 
 // Insert a row in the DB, it is assumed the table exists, you can do a
 // check before with HasTable()
-func (l *Loader) Insert(tableName string, primaryKey map[string]string, data map[string]string, ordinal uint64, reversibleBlockNum *uint64) error {
+func (l *Loader) Insert(tableName string, primaryKey map[string]string, data map[string]string, reversibleBlockNum *uint64) error {
 	uniqueID := createRowUniqueID(primaryKey)
 
 	if l.tracer.Enabled() {
@@ -59,7 +59,7 @@ func (l *Loader) Insert(tableName string, primaryKey map[string]string, data map
 		}
 	}
 
-	entry.Set(uniqueID, l.newInsertOperation(table, primaryKey, data, ordinal, reversibleBlockNum))
+	entry.Set(uniqueID, l.newInsertOperation(table, primaryKey, data, l.NextBatchOrdinal(), reversibleBlockNum))
 	l.entriesCount++
 	return nil
 }
@@ -101,7 +101,7 @@ func (l *Loader) GetPrimaryKey(tableName string, pk string) (map[string]string,
 
 // Upsert a row in the DB, it is assumed the table exists, you can do a
 // check before with HasTable().
-func (l *Loader) Upsert(tableName string, primaryKey map[string]string, data map[string]string, ordinal uint64, reversibleBlockNum *uint64) error {
+func (l *Loader) Upsert(tableName string, primaryKey map[string]string, data map[string]string, reversibleBlockNum *uint64) error {
 	if l.dialect.OnlyInserts() {
 		return fmt.Errorf("update operation is not supported by the current database")
 	}
@@ -147,7 +147,7 @@ func (l *Loader) Upsert(tableName string, primaryKey map[string]string, data map
 			l.logger.Debug("primary key entry already exist for table, merging columns together", zap.String("primary_key", uniqueID), zap.String("table_name", tableName))
 		}
 
-		op.mergeOperation(ordinal, data)
+		op.mergeOperation(data)
 		entry.Set(uniqueID, op)
 		return nil
 	} else {
@@ -165,13 +165,13 @@ func (l *Loader) Upsert(tableName string, primaryKey map[string]string, data map
 		}
 	}
 
-	entry.Set(uniqueID, l.newUpsertOperation(table, primaryKey, data, ordinal, reversibleBlockNum))
+	entry.Set(uniqueID, l.newUpsertOperation(table, primaryKey, data, l.NextBatchOrdinal(), reversibleBlockNum))
 	return nil
 }
 
 // Update a row in the DB, it is assumed the table exists, you can do a
 // check before with HasTable()
-func (l *Loader) Update(tableName string, primaryKey map[string]string, data map[string]string, ordinal uint64, reversibleBlockNum *uint64) error {
+func (l *Loader) Update(tableName string, primaryKey map[string]string, data map[string]string, reversibleBlockNum *uint64) error {
 	if l.dialect.OnlyInserts() {
 		return fmt.Errorf("update operation is not supported by the current database")
 	}
@@ -216,7 +216,7 @@ func (l *Loader) Update(tableName string, primaryKey map[string]string, data map
 			l.logger.Debug("primary key entry already exist for table, merging fields together", zap.String("primary_key", uniqueID), zap.String("table_name", tableName))
 		}
 
-		op.mergeOperation(ordinal, data)
+		op.mergeOperation(data)
 		entry.Set(uniqueID, op)
 		return nil
 	} else {
@@ -227,13 +227,13 @@ func (l *Loader) Update(tableName string, primaryKey map[string]string, data map
 		l.logger.Debug("primary key entry never existed for table, adding update operation", zap.String("primary_key", uniqueID), zap.String("table_name", tableName))
 	}
 
-	entry.Set(uniqueID, l.newUpdateOperation(table, primaryKey, data, ordinal, reversibleBlockNum))
+	entry.Set(uniqueID, l.newUpdateOperation(table, primaryKey, data, l.NextBatchOrdinal(), reversibleBlockNum))
 	return nil
 }
 
 // Delete a row in the DB, it is assumed the table exists, you can do a
 // check before with HasTable()
-func (l *Loader) Delete(tableName string, primaryKey map[string]string, ordinal uint64, reversibleBlockNum *uint64) error {
+func (l *Loader) Delete(tableName string, primaryKey map[string]string, reversibleBlockNum *uint64) error {
 	if l.dialect.OnlyInserts() {
 		return fmt.Errorf("delete operation is not supported by the current database")
 	}
@@ -274,6 +274,6 @@ func (l *Loader) Delete(tableName string, primaryKey map[string]string, ordinal
 		l.logger.Debug("adding deleting operation", zap.String("primary_key", uniqueID), zap.String("table_name", tableName))
 	}
 
-	entry.Set(uniqueID, l.newDeleteOperation(table, primaryKey, ordinal, reversibleBlockNum))
+	entry.Set(uniqueID, l.newDeleteOperation(table, primaryKey, l.NextBatchOrdinal(), reversibleBlockNum))
 	return nil
 }

db_changes/sinker/sinker.go

@@ -229,7 +229,7 @@ func (s *SQLSinker) HandleBlockScopedData(ctx context.Context, data *pbsubstream
 }
 
 func (s *SQLSinker) applyDatabaseChanges(dbChanges *pbdatabase.DatabaseChanges, blockNum, finalBlockNum uint64) error {
-	for ordinal, change := range dbChanges.TableChanges {
+	for _, change := range dbChanges.TableChanges {
 		if !s.loader.HasTable(change.Table) {
 			return fmt.Errorf(
 				"your Substreams sent us a change for a table named %s we don't know about on %s (available tables: %s)",
@@ -265,22 +265,22 @@ func (s *SQLSinker) applyDatabaseChanges(dbChanges *pbdatabase.DatabaseChanges,
 
 		switch change.Operation {
 		case pbdatabase.TableChange_OPERATION_CREATE:
-			err := s.loader.Insert(change.Table, primaryKeys, changes, uint64(ordinal), reversibleBlockNum)
+			err := s.loader.Insert(change.Table, primaryKeys, changes, reversibleBlockNum)
 			if err != nil {
 				return fmt.Errorf("database insert: %w", err)
 			}
 		case pbdatabase.TableChange_OPERATION_UPSERT:
-			err := s.loader.Upsert(change.Table, primaryKeys, changes, uint64(ordinal), reversibleBlockNum)
+			err := s.loader.Upsert(change.Table, primaryKeys, changes, reversibleBlockNum)
 			if err != nil {
 				return fmt.Errorf("database upsert: %w", err)
 			}
 		case pbdatabase.TableChange_OPERATION_UPDATE:
-			err := s.loader.Update(change.Table, primaryKeys, changes, uint64(ordinal), reversibleBlockNum)
+			err := s.loader.Update(change.Table, primaryKeys, changes, reversibleBlockNum)
 			if err != nil {
 				return fmt.Errorf("database update: %w", err)
 			}
 		case pbdatabase.TableChange_OPERATION_DELETE:
-			err := s.loader.Delete(change.Table, primaryKeys, uint64(ordinal), reversibleBlockNum)
+			err := s.loader.Delete(change.Table, primaryKeys, reversibleBlockNum)
 			if err != nil {
 				return fmt.Errorf("database delete: %w", err)
 			}

db_proto/sql/click_house/accumulator_inserter.go

@@ -188,6 +188,113 @@ func (i *AccumulatorInserter) insert(table string, values []any) error {
 			input.Append(value.([]byte))
 		case *proto.ColBool:
 			input.Append(value.(bool))
+		// Handle array column types
+		case *proto.ColArr[int32]:
+			if arr, ok := value.([]interface{}); ok {
+				int32Arr := make([]int32, len(arr))
+				for i, v := range arr {
+					int32Arr[i] = v.(int32)
+				}
+				input.Append(int32Arr)
+			} else {
+				panic(fmt.Sprintf("expected []interface{} for array column %s of table %s, got %T", colName, table, value))
+			}
+		case *proto.ColArr[int64]:
+			if arr, ok := value.([]interface{}); ok {
+				int64Arr := make([]int64, len(arr))
+				for i, v := range arr {
+					int64Arr[i] = v.(int64)
+				}
+				input.Append(int64Arr)
+			} else {
+				panic(fmt.Sprintf("expected []interface{} for array column %s of table %s, got %T", colName, table, value))
+			}
+		case *proto.ColArr[uint32]:
+			if arr, ok := value.([]interface{}); ok {
+				uint32Arr := make([]uint32, len(arr))
+				for i, v := range arr {
+					uint32Arr[i] = v.(uint32)
+				}
+				input.Append(uint32Arr)
+			} else {
+				panic(fmt.Sprintf("expected []interface{} for array column %s of table %s, got %T", colName, table, value))
+			}
+		case *proto.ColArr[uint64]:
+			if arr, ok := value.([]interface{}); ok {
+				uint64Arr := make([]uint64, len(arr))
+				for i, v := range arr {
+					uint64Arr[i] = v.(uint64)
+				}
+				input.Append(uint64Arr)
+			} else {
+				panic(fmt.Sprintf("expected []interface{} for array column %s of table %s, got %T", colName, table, value))
+			}
+		case *proto.ColArr[float32]:
+			if arr, ok := value.([]interface{}); ok {
+				float32Arr := make([]float32, len(arr))
+				for i, v := range arr {
+					float32Arr[i] = v.(float32)
+				}
+				input.Append(float32Arr)
+			} else {
+				panic(fmt.Sprintf("expected []interface{} for array column %s of table %s, got %T", colName, table, value))
+			}
+		case *proto.ColArr[float64]:
+			if arr, ok := value.([]interface{}); ok {
+				float64Arr := make([]float64, len(arr))
+				for i, v := range arr {
+					float64Arr[i] = v.(float64)
+				}
+				input.Append(float64Arr)
+			} else {
+				panic(fmt.Sprintf("expected []interface{} for array column %s of table %s, got %T", colName, table, value))
+			}
+		case *proto.ColArr[bool]:
+			if arr, ok := value.([]interface{}); ok {
+				boolArr := make([]bool, len(arr))
+				for i, v := range arr {
+					boolArr[i] = v.(bool)
+				}
+				input.Append(boolArr)
+			} else {
+				panic(fmt.Sprintf("expected []interface{} for array column %s of table %s, got %T", colName, table, value))
+			}
+		case *proto.ColArr[string]:
+			if arr, ok := value.([]interface{}); ok {
+				stringArr := make([]string, len(arr))
+				for i, v := range arr {
+					stringArr[i] = v.(string)
+				}
+				input.Append(stringArr)
+			} else {
+				panic(fmt.Sprintf("expected []interface{} for array column %s of table %s, got %T", colName, table, value))
+			}
+		case *proto.ColArr[[]byte]:
+			if arr, ok := value.([]interface{}); ok {
+				bytesArr := make([][]byte, len(arr))
+				for i, v := range arr {
+					bytesArr[i] = v.([]byte)
+				}
+				input.Append(bytesArr)
+			} else {
+				panic(fmt.Sprintf("expected []interface{} for array column %s of table %s, got %T", colName, table, value))
+			}
+		case *proto.ColArr[time.Time]:
+			if arr, ok := value.([]interface{}); ok {
+				timeArr := make([]time.Time, len(arr))
+				for i, v := range arr {
+					if t, ok := v.(*timestamppb.Timestamp); ok {
+						timeArr[i] = t.AsTime()
+					} else if t, ok := v.(time.Time); ok {
+						timeArr[i] = t
+					} else {
+						panic(fmt.Sprintf("unknown time type %T in array for column %s of table %s", v, colName, table))
+					}
+				}
+				input.Append(timeArr)
+			} else {
+				panic(fmt.Sprintf("expected []interface{} for array column %s of table %s, got %T", colName, table, value))
+			}
 		default:
 			panic(fmt.Sprintf("unknown input type %T for column %s of table %s", input, colName, table))
 		}

db_proto/sql/click_house/dialect.go

@@ -114,13 +114,6 @@ func (d *DialectClickHouse) createTable(table *schema.Table) error {
 
 		fieldName := f.Name
 
-		switch {
-		case f.IsRepeated:
-			continue
-		case f.IsMessage:
-		case f.ForeignKey != nil:
-		}
-
 		fieldType := MapFieldType(f.FieldDescriptor)
 		sb.WriteString(fmt.Sprintf("%s %s", fieldName, fieldType))
 		sb.WriteString(",")