diff --git a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableTableModifyRule.java b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableTableModifyRule.java
index 4e111fe2248a..651f008f8d0b 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableTableModifyRule.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableTableModifyRule.java
@@ -17,17 +17,66 @@
 package org.apache.calcite.adapter.enumerable;
 
 import org.apache.calcite.plan.Convention;
+import org.apache.calcite.plan.RelOptCluster;
 import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.prepare.RelOptTableImpl;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.convert.ConverterRule;
 import org.apache.calcite.rel.core.TableModify;
 import org.apache.calcite.rel.logical.LogicalTableModify;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.rex.RexUtil;
 import org.apache.calcite.schema.ModifiableTable;
+import org.apache.calcite.sql.fun.SqlInternalOperators;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.sql.type.SqlTypeUtil;
 
 import org.checkerframework.checker.nullness.qual.Nullable;
 
-/** Planner rule that converts a {@link LogicalTableModify} to an {@link EnumerableTableModify}.
- * You may provide a custom config to convert other nodes that extend {@link TableModify}.
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+/** Planner rule that converts a {@link LogicalTableModify} to an
+ * {@link EnumerableTableModify}.
+ *
+ * <p>For INSERT and UPDATE, the rule adds assignment checks of the following
+ * form:
+ *
+ * <blockquote><pre>{@code
+ * LogicalTableModify
+ *   input
+ *
+ * EnumerableTableModify
+ *   EnumerableCalc(
+ *     condition=[
+ *       AND(
+ *         $THROW_UNLESS(
+ *           IS_NOT_FALSE(
+ *             LESS_THAN_OR_EQUAL(CHAR_LENGTH($0), targetPrecision)), ...),
+ *         $THROW_UNLESS(
+ *           IS_NOT_FALSE(
+ *             LESS_THAN_OR_EQUAL(OCTET_LENGTH($1), targetPrecision)), ...))],
+ *     projects=[
+ *       $0,
+ *       $1,
+ *       CAST($2 AS targetType)])
+ *     enumerable input
+ * }</pre></blockquote>
+ *
+ * <p>Character and binary values are checked by length because narrowing casts
+ * may truncate them. Exact numeric values use a target cast, which throws on
+ * overflow.
+ *
+ * <p>You may provide a custom config to convert other nodes that extend
+ * {@link TableModify}.
  *
  * @see EnumerableRules#ENUMERABLE_TABLE_MODIFICATION_RULE */
 public class EnumerableTableModifyRule extends ConverterRule {
@@ -44,6 +93,7 @@ protected EnumerableTableModifyRule(Config config) {
 
   @Override public @Nullable RelNode convert(RelNode rel) {
     final TableModify modify = (TableModify) rel;
+    final RelOptCluster cluster = modify.getCluster();
     final ModifiableTable modifiableTable =
         modify.getTable().unwrap(ModifiableTable.class);
     if (modifiableTable == null) {
@@ -51,11 +101,81 @@ protected EnumerableTableModifyRule(Config config) {
     }
     final RelTraitSet traitSet =
         modify.getTraitSet().replace(EnumerableConvention.INSTANCE);
+    RelNode input = convert(modify.getInput(), traitSet);
+    if (modify.isInsert() || modify.isUpdate()) {
+      // INSERT assigns stored columns; UPDATE assigns columns in the SET list.
+      RelDataType assignmentType = modify.isInsert()
+          ? RelOptTableImpl.realRowType(modify.getTable())
+          : modify.getCatalogReader().createTypeFromProjection(
+              modify.getTable().getRowType(),
+              requireNonNull(modify.getUpdateColumnList(), "updateColumnList"));
+      if (modify.isFlattened()) {
+        // TableModify flattens its input, so flatten the target fields too.
+        assignmentType =
+            SqlTypeUtil.flattenRecordType(cluster.getTypeFactory(), assignmentType, null);
+      }
+
+      final RexBuilder rexBuilder = cluster.getRexBuilder();
+      final List<RexNode> projects =
+          new ArrayList<>(rexBuilder.identityProjects(input.getRowType()));
+      final List<RexNode> checks = new ArrayList<>();
+      // UPDATE appends SET values to the old row; INSERT has only new values.
+      final int assignmentOffset = projects.size() - assignmentType.getFieldCount();
+      for (RelDataTypeField field : assignmentType.getFieldList()) {
+        final int sourceOrdinal = assignmentOffset + field.getIndex();
+        final RexNode source = projects.get(sourceOrdinal);
+        final RelDataType targetType = field.getType();
+        final SqlTypeName targetName = targetType.getSqlTypeName();
+        if (SqlTypeUtil.inCharOrBinaryFamilies(targetType)) {
+          if (targetType.getPrecision() < 0) {
+            continue;
+          }
+          // Check character and binary lengths because their casts may truncate.
+          final RexNode length =
+              rexBuilder.makeCall(SqlTypeUtil.inCharFamily(targetType)
+                  ? SqlStdOperatorTable.CHAR_LENGTH
+                  : SqlStdOperatorTable.OCTET_LENGTH,
+              source);
+          final RexNode fits =
+              rexBuilder.makeCall(SqlStdOperatorTable.LESS_THAN_OR_EQUAL, length,
+              rexBuilder.makeExactLiteral(
+                  BigDecimal.valueOf(targetType.getPrecision())));
+          // IS_NOT_FALSE lets NULL pass this length check.
+          final RexNode valid =
+              rexBuilder.makeCall(SqlStdOperatorTable.IS_NOT_FALSE, fits);
+          checks.add(
+              rexBuilder.makeCall(SqlInternalOperators.THROW_UNLESS,
+              valid, rexBuilder.makeLiteral("Value exceeds precision "
+                  + targetType.getPrecision() + " of "
+                  + targetType.getFullTypeString())));
+        } else {
+          if (targetName == SqlTypeName.DECIMAL) {
+            // A runtime BigDecimal may exceed its declared precision.
+            if (targetType.getPrecision() < 0 || targetType.getScale() < 0) {
+              continue;
+            }
+          } else if (!SqlTypeUtil.isExactNumeric(targetType)
+              || source.getType().getSqlTypeName() == targetName) {
+            continue;
+          }
+          // Exact numeric casts reject overflow and produce the target value.
+          projects.set(sourceOrdinal, rexBuilder.makeCast(targetType, source));
+        }
+      }
+      if (!checks.isEmpty() || !RexUtil.isIdentity(projects, input.getRowType())) {
+        // RexProgram has one condition, so combine all length checks.
+        input =
+            EnumerableCalc.create(
+                input, RexProgram.create(input.getRowType(), projects,
+                RexUtil.composeConjunction(rexBuilder, checks, true),
+                input.getRowType().getFieldNames(), rexBuilder));
+      }
+    }
     return new EnumerableTableModify(
-        modify.getCluster(), traitSet,
+        cluster, traitSet,
         modify.getTable(),
         modify.getCatalogReader(),
-        convert(modify.getInput(), traitSet),
+        input,
         modify.getOperation(),
         modify.getUpdateColumnList(),
         modify.getSourceExpressionList(),
diff --git a/server/src/test/java/org/apache/calcite/test/ServerTest.java b/server/src/test/java/org/apache/calcite/test/ServerTest.java
index 0be7fb16efa1..ab7bbd9d5271 100644
--- a/server/src/test/java/org/apache/calcite/test/ServerTest.java
+++ b/server/src/test/java/org/apache/calcite/test/ServerTest.java
@@ -16,11 +16,13 @@
  */
 package org.apache.calcite.test;
 
+import org.apache.calcite.avatica.util.ByteString;
 import org.apache.calcite.config.CalciteConnectionProperty;
 import org.apache.calcite.jdbc.CalciteConnection;
 import org.apache.calcite.jdbc.CalcitePrepare;
 import org.apache.calcite.schema.Function;
 import org.apache.calcite.schema.FunctionParameter;
+import org.apache.calcite.schema.ModifiableTable;
 import org.apache.calcite.server.DdlExecutorImpl;
 import org.apache.calcite.server.ServerDdlExecutor;
 import org.apache.calcite.sql.SqlNode;
@@ -49,6 +51,7 @@
 import java.sql.Statement;
 import java.sql.Struct;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 
 import static org.apache.calcite.test.Matchers.isLinux;
@@ -56,6 +59,7 @@
 import static org.hamcrest.CoreMatchers.containsString;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.CoreMatchers.nullValue;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
@@ -64,6 +68,8 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
+import static java.util.Objects.requireNonNull;
+
 /**
  * Unit tests for server and DDL.
  */
@@ -82,6 +88,21 @@ static Connection connect() throws SQLException {
             .build());
   }
 
+  private static void assertFails(Statement statement, String sql, String message) {
+    final SQLException e =
+        assertThrows(SQLException.class, () -> statement.executeUpdate(sql));
+    assertThat(e.getMessage(), containsString(message));
+  }
+
+  @SuppressWarnings("unchecked")
+  private static Collection<Object> rows(Connection connection, String tableName)
+      throws SQLException {
+    return (Collection<Object>) (
+        (ModifiableTable) requireNonNull(
+        connection.unwrap(CalciteConnection.class).getRootSchema().tables().get(tableName)))
+        .getModifiableCollection();
+  }
+
   /** Contains calls to all overloaded {@code execute} methods in
    * {@link DdlExecutorImpl} to silence warnings that these methods are not
    * called. (They are, not from this test, but via reflection.) */
@@ -150,6 +171,143 @@ static Connection connect() throws SQLException {
     }
   }
 
+  /** Test case for
+   * <a href="https://issues.apache.org/jira/browse/CALCITE-7627">[CALCITE-7627]
+   * Enumerable DML should reject assignments that may lose data</a>. */
+  @Test void testEnumerableCharacterAssignment() throws Exception {
+    try (Connection c = connect();
+         Statement s = c.createStatement()) {
+      s.execute("create table dept (deptno integer not null, name varchar(10))");
+
+      // Reject an overlength VARCHAR value.
+      assertFails(s, "insert into dept values (10, 'Engineering')",
+          "exceeds precision 10 of VARCHAR(10)");
+      // Keep a cast already present in the input plan.
+      assertThat(
+          s.executeUpdate(
+          "insert into dept values (10, upper('Engineering'))"), is(1));
+
+      // Count trailing spaces toward VARCHAR precision.
+      assertFails(s, "insert into dept values (10, '1234567890  ')",
+          "exceeds precision 10 of VARCHAR(10)");
+
+      // Keep explicit CAST semantics.
+      assertThat(
+          s.executeUpdate("insert into dept values "
+          + "(20, cast('Engineering' as varchar(10)))"), is(1));
+      // Accept NULL.
+      assertThat(s.executeUpdate("insert into dept values (40, null)"), is(1));
+
+      // Check UPDATE assignments.
+      assertFails(s, "update dept set name = 'Engineering' where deptno = 20",
+          "exceeds precision 10 of VARCHAR(10)");
+
+      // Check runtime values even when source and target types match.
+      s.execute("create table same_names (name varchar(10))");
+      rows(c, "SAME_NAMES").add("Engineering");
+      assertFails(s, "insert into dept select 30, name from same_names",
+          "exceeds precision 10 of VARCHAR(10)");
+
+      // Store values produced by existing casts unchanged.
+      try (ResultSet r =
+          s.executeQuery("select name from dept where name is not null order by deptno")) {
+        assertThat(r.next(), is(true));
+        assertThat(r.getString(1), is("ENGINEERIN"));
+        assertThat(r.next(), is(true));
+        assertThat(r.getString(1), is("Engineerin"));
+        assertThat(r.next(), is(false));
+      }
+
+      // Reject an overlength CHAR value.
+      s.execute("create table fixed_names (name char(5))");
+      s.execute("create table same_fixed_names (name char(5))");
+      rows(c, "SAME_FIXED_NAMES").add("abcdef");
+      assertFails(s, "insert into fixed_names select * from same_fixed_names",
+          "exceeds precision 5 of CHAR(5)");
+    }
+  }
+
+  @Test void testEnumerableBinaryAssignment() throws Exception {
+    try (Connection c = connect();
+         Statement s = c.createStatement()) {
+      s.execute("create table bytes (b binary(4), vb varbinary(2))");
+      // Accept values within BINARY and VARBINARY bounds.
+      assertThat(
+          s.executeUpdate(
+          "insert into bytes values (x'0102', x'0304')"), is(1));
+
+      // Reject a BINARY value that exceeds its target precision.
+      s.execute("create table same_bytes (b binary(4), vb varbinary(2))");
+      rows(c, "SAME_BYTES").add(new Object[] {
+          ByteString.of("0102030405", 16), ByteString.of("06", 16)});
+      assertFails(s, "insert into bytes select * from same_bytes",
+          "exceeds precision 4 of BINARY(4)");
+
+      // Keep accepted binary values unchanged.
+      try (ResultSet r = s.executeQuery("select b, vb from bytes")) {
+        assertThat(r.next(), is(true));
+        assertArrayEquals(new byte[] {1, 2}, r.getBytes(1));
+        assertArrayEquals(new byte[] {3, 4}, r.getBytes(2));
+        assertThat(r.next(), is(false));
+      }
+    }
+  }
+
+  @Test void testEnumerableExactNumericAssignment() throws Exception {
+    try (Connection c = connect();
+         Statement s = c.createStatement()) {
+      s.execute("create table numbers (d decimal(5, 2), i integer)");
+      // Keep input-plan DECIMAL rounding.
+      assertThat(s.executeUpdate("insert into numbers values (12.345, 1)"), is(1));
+
+      s.execute("create table wide_numbers (i bigint)");
+      s.executeUpdate("insert into wide_numbers values (42), (null), (2147483648)");
+      // Convert an in-range BIGINT value to INTEGER.
+      assertThat(
+          s.executeUpdate(
+          "insert into numbers select 1, i from wide_numbers where i = 42"), is(1));
+      // Accept a NULL numeric value.
+      assertThat(
+          s.executeUpdate(
+          "insert into numbers select 1, i from wide_numbers where i is null"), is(1));
+      // Reject an out-of-range BIGINT value.
+      assertFails(s, "insert into numbers select 1, i from wide_numbers "
+              + "where i = 2147483648",
+          "Value 2147483648 out of range");
+
+      s.execute("create table same_numbers (d decimal(5, 2))");
+      final Collection<Object> sameNumbers = rows(c, "SAME_NUMBERS");
+      // Apply target scale even when DECIMAL types match.
+      sameNumbers.add(new BigDecimal("1.234"));
+      assertThat(
+          s.executeUpdate(
+          "insert into numbers select d, 2 from same_numbers"), is(1));
+      sameNumbers.clear();
+      // Check runtime DECIMAL values even when declared types match.
+      sameNumbers.add(new BigDecimal("1000.00"));
+      assertFails(s, "insert into numbers select d, 1 from same_numbers",
+          "Value 1000.00 cannot be represented as a DECIMAL(5, 2)");
+
+      // Read back the converted numeric values.
+      try (ResultSet r =
+          s.executeQuery("select d, i from numbers order by i nulls last")) {
+        assertThat(r.next(), is(true));
+        assertThat(r.getBigDecimal(1), is(new BigDecimal("12.35")));
+        assertThat(r.getInt(2), is(1));
+        assertThat(r.next(), is(true));
+        assertThat(r.getBigDecimal(1), is(new BigDecimal("1.23")));
+        assertThat(r.getInt(2), is(2));
+        assertThat(r.next(), is(true));
+        assertThat(r.getBigDecimal(1), is(new BigDecimal("1.00")));
+        assertThat(r.getInt(2), is(42));
+        assertThat(r.next(), is(true));
+        assertThat(r.getBigDecimal(1), is(new BigDecimal("1.00")));
+        assertThat(r.getObject(2), nullValue());
+        assertThat(r.next(), is(false));
+      }
+    }
+  }
+
   @Test void testUpdateDuplicateRows() throws Exception {
     try (Connection c = connect();
          Statement s = c.createStatement()) {
diff --git a/server/src/test/resources/sql/table.iq b/server/src/test/resources/sql/table.iq
index b40f7b0be367..5597dbbd9f4a 100755
--- a/server/src/test/resources/sql/table.iq
+++ b/server/src/test/resources/sql/table.iq
@@ -323,7 +323,8 @@ insert into t (i, k) values ('abcde', 'de ');
 
 !update
 EnumerableTableModify(table=[[T]], operation=[INSERT], flattened=[false])
-  EnumerableValues(tuples=[[{ 'abcde', 'de ' }]])
+  EnumerableCalc(expr#0..1=[{inputs}], expr#2=[CHAR_LENGTH($t0)], expr#3=[5], expr#4=[<=($t2, $t3)], expr#5=['Value exceeds precision 5 of VARCHAR(5)'], expr#6=[$THROW_UNLESS($t4, $t5)], expr#7=[CHAR_LENGTH($t1)], expr#8=[3], expr#9=[<=($t7, $t8)], expr#10=['Value exceeds precision 3 of VARCHAR(3)'], expr#11=[$THROW_UNLESS($t9, $t10)], expr#12=[AND($t6, $t11)], proj#0..1=[{exprs}], $condition=[$t12])
+    EnumerableValues(tuples=[[{ 'abcde', 'de ' }]])
 !plan
 
 insert into t (k, i) values ('de ', 'abcde');
@@ -331,7 +332,7 @@ insert into t (k, i) values ('de ', 'abcde');
 
 !update
 EnumerableTableModify(table=[[T]], operation=[INSERT], flattened=[false])
-  EnumerableCalc(expr#0..1=[{inputs}], I=[$t1], K=[$t0])
+  EnumerableCalc(expr#0..1=[{inputs}], expr#2=[CHAR_LENGTH($t1)], expr#3=[5], expr#4=[<=($t2, $t3)], expr#5=['Value exceeds precision 5 of VARCHAR(5)'], expr#6=[$THROW_UNLESS($t4, $t5)], expr#7=[CHAR_LENGTH($t0)], expr#8=[3], expr#9=[<=($t7, $t8)], expr#10=['Value exceeds precision 3 of VARCHAR(3)'], expr#11=[$THROW_UNLESS($t9, $t10)], expr#12=[AND($t6, $t11)], I=[$t1], K=[$t0], $condition=[$t12])
     EnumerableValues(tuples=[[{ 'de ', 'abcde' }]])
 !plan