Merge remote-tracking branch 'origin/236-backup' into 236-backup

backend/loaddata.go

@@ -117,6 +117,16 @@ func (db *DuckBuilder) executeLoadData(ctx *sql.Context, insert *plan.InsertInto
 	b.Grow(256)
 
 	keyless := sql.IsKeyless(dst.Schema())
+	// TODO(fan): This is an ugly hack for MySQL Shell's utilities to work properly.
+	if !keyless {
+		if t, ok := dst.(*catalog.Table); ok {
+			// Replicated tables do not have physical primary keys.
+			// Their logical primary keys are fake and should not be used in INSERT INTO statements.
+			// https://github.com/apecloud/myduckserver/issues/272
+			keyless = t.ExtraTableInfo().Replicated
+		}
+	}
+
 	b.WriteString("INSERT")
 	if load.IsIgnore && !keyless {
 		b.WriteString(" OR IGNORE")

catalog/database.go

@@ -170,11 +170,11 @@ func (d *Database) createAllTable(ctx *sql.Context, name string, schema sql.Prim
 		primaryKeys = append(primaryKeys, schema.Schema[pkord].Name)
 	}
 
+	withoutIndex := isIndexCreationDisabled(ctx)
+
 	// https://github.com/apecloud/myduckserver/issues/272
-	if !(mycontext.IsReplicationQuery(ctx) && configuration.IsReplicationWithoutIndex()) {
-		if len(primaryKeys) > 0 {
-			b.WriteString(fmt.Sprintf(", PRIMARY KEY (%s)", strings.Join(primaryKeys, ", ")))
-		}
+	if len(primaryKeys) > 0 && !withoutIndex {
+		b.WriteString(fmt.Sprintf(", PRIMARY KEY (%s)", strings.Join(primaryKeys, ", ")))
 	}
 
 	b.WriteString(")")
@@ -183,7 +183,7 @@ func (d *Database) createAllTable(ctx *sql.Context, name string, schema sql.Prim
 	b.WriteString(fmt.Sprintf(
 		"; COMMENT ON TABLE %s IS '%s'",
 		fullTableName,
-		NewCommentWithMeta(comment, ExtraTableInfo{schema.PkOrdinals}).Encode(),
+		NewCommentWithMeta(comment, ExtraTableInfo{schema.PkOrdinals, withoutIndex}).Encode(),
 	))
 
 	// Add column comments
@@ -205,6 +205,23 @@ func (d *Database) createAllTable(ctx *sql.Context, name string, schema sql.Prim
 	return nil
 }
 
+func isIndexCreationDisabled(ctx *sql.Context) bool {
+	if !configuration.IsReplicationWithoutIndex() {
+		return false
+	}
+	if mycontext.IsReplicationQuery(ctx) {
+		return true
+	}
+	_, vv, ok := sql.SystemVariables.GetGlobal("replica_is_loading_snapshot")
+	if !ok {
+		return false
+	}
+	if b, ok := vv.(int8); ok {
+		return b != 0
+	}
+	return false
+}
+
 // CreateTable implements sql.TableCreator.
 func (d *Database) CreateTable(ctx *sql.Context, name string, schema sql.PrimaryKeySchema, collation sql.CollationID, comment string) error {
 	d.mu.Lock()

catalog/table.go

@@ -8,7 +8,6 @@ import (
 
 	"github.com/apecloud/myduckserver/adapter"
 	"github.com/apecloud/myduckserver/configuration"
-	"github.com/apecloud/myduckserver/mycontext"
 	"github.com/dolthub/go-mysql-server/sql"
 	"github.com/dolthub/go-mysql-server/sql/expression"
 	"github.com/marcboeker/go-duckdb"
@@ -25,6 +24,7 @@ type Table struct {
 
 type ExtraTableInfo struct {
 	PkOrdinals []int
+	Replicated bool
 }
 
 type ColumnInfo struct {
@@ -362,7 +362,7 @@ func (t *Table) CreateIndex(ctx *sql.Context, indexDef sql.IndexDef) error {
 	defer t.mu.Unlock()
 
 	// https://github.com/apecloud/myduckserver/issues/272
-	if mycontext.IsReplicationQuery(ctx) && configuration.IsReplicationWithoutIndex() {
+	if isIndexCreationDisabled(ctx) {
 		return nil
 	}
 

configuration/env.go

@@ -11,7 +11,7 @@ const (
 
 func IsReplicationWithoutIndex() bool {
 	switch strings.ToLower(os.Getenv(replicationWithoutIndex)) {
-	case "", "t", "1", "true":
+	case "", "y", "t", "1", "on", "yes", "true":
 		return true
 	}
 	return false

devtools/replica-setup-mysql/prepare.sh

@@ -17,4 +17,5 @@ echo "Setting local_infile and server_id..."
 mysqlsh --sql --host=${MYDUCK_HOST} --port=${MYDUCK_PORT}  --user=root --no-password <<EOF
 SET GLOBAL local_infile = 1;
 SET GLOBAL server_id = ${MYDUCK_SERVER_ID};
+SET GLOBAL replica_is_loading_snapshot = ON;
 EOF

devtools/replica-setup-mysql/snapshot.sh

@@ -36,7 +36,17 @@ echo "Thread count set to: $THREAD_COUNT"
 
 echo "Copying data from MySQL to MyDuck..."
 # Run mysqlsh command and capture the output
-output=$(mysqlsh --uri "$SOURCE_DSN" $SOURCE_NO_PASSWORD_OPTION -- util copy-instance "mysql://${MYDUCK_USER}:${MYDUCK_PASSWORD}@${MYDUCK_HOST}:${MYDUCK_PORT}" --users false --consistent false --ignore-existing-objects true --handle-grant-errors ignore --threads $THREAD_COUNT --bytesPerChunk 256M --ignore-version true)
+output=$(mysqlsh --uri "$SOURCE_DSN" $SOURCE_NO_PASSWORD_OPTION -- util copy-instance "mysql://${MYDUCK_USER}:${MYDUCK_PASSWORD}@${MYDUCK_HOST}:${MYDUCK_PORT}" \
+    --users false \
+    --consistent false \
+    --ignore-existing-objects true \
+    --handle-grant-errors ignore \
+    --threads $THREAD_COUNT \
+    --bytesPerChunk 256M \
+    --ignore-version true \
+    --load-indexes false \
+    --defer-table-indexes all \
+)
 
 if [[ $GTID_MODE == "ON" ]]; then
     # Extract the EXECUTED_GTID_SET from this output:
@@ -71,4 +81,9 @@ else
 fi
 
 
-echo "Snapshot completed successfully."
+echo "Snapshot completed successfully."
+
+echo "Reset replica_is_loading_snapshot..."
+mysqlsh --sql --host=${MYDUCK_HOST} --port=${MYDUCK_PORT}  --user=root --no-password <<EOF
+SET GLOBAL replica_is_loading_snapshot = OFF;
+EOF

replica/options.go

@@ -48,5 +48,14 @@ func RegisterReplicaOptions(options *ReplicaOptions) {
 			Type:              types.NewSystemStringType("report_password"),
 			Default:           options.ReportPassword,
 		},
+		// MyDuck-specific system variables
+		&sql.MysqlSystemVariable{
+			Name:              "replica_is_loading_snapshot",
+			Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Global),
+			Dynamic:           true,
+			SetVarHintApplies: false,
+			Type:              types.NewSystemBoolType("replica_is_loading_snapshot"),
+			Default:           false,
+		},
 	})
 }