fix!: Split max index length between PKs (#13569)

erezrokah · web-flow · commit b0ad9d4c62d4 · 2023-09-04T07:10:16.000Z
#### Summary Fixes #13486. The max length of an index is either `3072` bytes or `767` bytes. Since we use `utf8mb4` which can use up to 4 bytes per char, we max out at either `768` or `191` chars (the number in bytes divided by `4`). The current code allocates `191` chars per PK which is OK if a table has 4 PKs since `191*4= 764`, but fails if a table has more than that. See list of tables [here](#13486 (comment)) The solution is to get the max number of bytes based on the row format, and evenly distribute the max length between PKs. This is still not the best solution as we assign the same number of bytes to a UUID and a `text` column (for example CQId is `UUID`), but I don't think it's very common to have one UUID PK and another `text` PK and enough other PKs so the prefix is too short. <!--
diff --git a/plugins/destination/mysql/client/client.go b/plugins/destination/mysql/client/client.go
@@ -32,6 +32,8 @@ type Client struct {
 	writer        *batchwriter.BatchWriter
 	serverType    ServerType
 	serverVersion string
+
+	maxIndexLength int
 }
 
 func New(ctx context.Context, logger zerolog.Logger, spec []byte, _ plugin.NewClientOptions) (plugin.Client, error) {
@@ -77,6 +79,8 @@ func New(ctx context.Context, logger zerolog.Logger, spec []byte, _ plugin.NewCl
 		return nil, err
 	}
 
+	c.setMaxIndexLength(ctx)
+
 	return c, nil
 }
 
@@ -118,6 +122,26 @@ func (c *Client) getVersion(ctx context.Context) error {
 	return nil
 }
 
+func (c *Client) setMaxIndexLength(ctx context.Context) {
+	const maxIndexLengthInBytes = 3072
+	row := c.db.QueryRowContext(ctx, "show variables like 'innodb_default_row_format'")
+	var varName sql.NullString
+	var rowFormat sql.NullString
+	err := row.Scan(&varName, &rowFormat)
+	if err != nil {
+		c.logger.Warn().Err(err).Msgf("failed to detect max index length, using default value of %d bytes", maxIndexLengthInBytes)
+	}
+
+	// In MySQL >= 5.7 the max PK length is 3072 bytes for dynamic or compressed row format, and 767 bytes for redundant or compact row format.
+	// We need to divide the max length in bytes by 4, since we use utf8mb4 charset, which can take up to 4 bytes per character.
+	switch rowFormat.String {
+	case "redundant", "compact":
+		c.maxIndexLength = 767 / 4
+	default:
+		c.maxIndexLength = maxIndexLengthInBytes / 4
+	}
+}
+
 func (c *Client) Close(ctx context.Context) error {
 	if err := c.writer.Close(ctx); err != nil {
 		_ = c.db.Close()
diff --git a/plugins/destination/mysql/client/schema.go b/plugins/destination/mysql/client/schema.go
@@ -13,20 +13,13 @@ func identifier(name string) string {
 	return fmt.Sprintf("`%s`", name)
 }
 
-const maxPrefixLength = 191
-
 const columnQuery = `SELECT 
 cols.COLUMN_NAME,
 COLUMN_TYPE,
 IS_NULLABLE,
-constraint_type,
-sub_part
+constraint_type
 FROM
 INFORMATION_SCHEMA.COLUMNS AS cols
-	LEFT JOIN  information_schema.STATISTICS as stats on
-		cols.table_schema = stats.table_schema and
-		cols.TABLE_NAME = stats.table_name and
-		cols.COLUMN_NAME = stats.column_name and index_name = 'PRIMARY'
 	LEFT JOIN
 (SELECT 
 	tc.constraint_schema,
@@ -62,8 +55,7 @@ func (c *Client) getTableColumns(ctx context.Context, tableName string) ([]schem
 		var typ string
 		var nullable string
 		var constraintType *string
-		var subpart *int
-		if err := rows.Scan(&name, &typ, &nullable, &constraintType, &subpart); err != nil {
+		if err := rows.Scan(&name, &typ, &nullable, &constraintType); err != nil {
 			return nil, err
 		}
 
@@ -72,10 +64,6 @@ func (c *Client) getTableColumns(ctx context.Context, tableName string) ([]schem
 		if constraintType != nil {
 			primaryKey = strings.Contains(*constraintType, "PRIMARY KEY")
 		}
-		// subpart only non nil for pks on blob/text columns
-		if subpart != nil && *subpart != maxPrefixLength {
-			primaryKey = false
-		}
 		columns = append(columns, schema.Column{
 			Name:       name,
 			Type:       schemaType,
@@ -172,15 +160,15 @@ func (c *Client) createTable(ctx context.Context, table *schema.Table) error {
 	if len(primaryKeysIndices) > 0 {
 		builder.WriteString(",\n  ")
 		builder.WriteString(" PRIMARY KEY (")
+		lengthPerPk := c.maxIndexLength / len(primaryKeysIndices)
 		for i, pk := range primaryKeysIndices {
 			column := table.Columns[pk]
 			builder.WriteString(identifier(column.Name))
 			sqlType := arrowTypeToMySqlStr(column.Type)
 			if sqlType == "blob" || sqlType == "text" {
 				// `blob/text` SQL types require specifying prefix length to use for the primary key
 				// https://dev.mysql.com/doc/refman/8.0/en/innodb-limits.html
-				// The index key prefix length limit is 767 bytes for InnoDB tables that use the REDUNDANT or COMPACT row format. For example, you might hit this limit with a column prefix index of more than 191 characters on a TEXT or VARCHAR column, assuming a utf8mb4 character set and the maximum of 4 bytes for each character.
-				builder.WriteString("(" + strconv.Itoa(maxPrefixLength) + ")")
+				builder.WriteString("(" + strconv.Itoa(lengthPerPk) + ")")
 			}
 			if i < len(primaryKeysIndices)-1 {
 				builder.WriteString(", ")
diff --git a/plugins/destination/mysql/client/write.go b/plugins/destination/mysql/client/write.go
@@ -60,16 +60,17 @@ func (c *Client) writeResources(ctx context.Context, query string, table *schema
 			return err
 		}
 
-		// log a warning that a blob or text field that is a PK has more than 191 characters
 		for _, record := range transformedRecords {
 			for _, truncatablePKIndex := range pks {
-				if len(record[truncatablePKIndex].(string)) > maxPrefixLength {
+				// log a warning that a blob or text field that is a PK has more than the limit
+				lengthPerPk := c.maxIndexLength / len(pks)
+				if len(record[truncatablePKIndex].(string)) > lengthPerPk {
 					indexes := table.PrimaryKeysIndexes()
 					pkValues := make(map[string]any, len(indexes))
 					for i, pkIndex := range indexes {
 						pkValues[table.Columns[pkIndex].Name] = record[i]
 					}
-					c.logger.Debug().Any("pk_values", pkValues).Msgf("record contains a primary key that is longer than MySQL can handle. only the first %d will be included in the index", maxPrefixLength)
+					c.logger.Debug().Any("pk_values", pkValues).Msgf("record contains a primary key that is longer than MySQL can handle. only the first %d will be included in the index", lengthPerPk)
 					tablesWithTruncation[table.Name] = true
 					break
 				}
