public KsqlGenericRecord build(
      final List<ColumnName> columnNames,
      final List<Expression> expressions,
      final LogicalSchema schema,
      final DataSourceType dataSourceType
  ) {
    final List<ColumnName> columns = columnNames.isEmpty()
        ? implicitColumns(schema)
        : columnNames;

    if (columns.size() != expressions.size()) {
      throw new KsqlException(
          "Expected a value for each column."
              + " Expected Columns: " + columnNames
              + ". Got " + expressions);
    }

    final LogicalSchema schemaWithPseudoColumns = withPseudoColumns(schema);
    for (ColumnName col : columns) {

      if (!schemaWithPseudoColumns.findColumn(col).isPresent()) {
        throw new KsqlException("Column name " + col + " does not exist.");
      }

      if (SystemColumns.isDisallowedForInsertValues(col)) {
        throw new KsqlException("Inserting into column " + col + " is not allowed.");
      }
    }

    final Map<ColumnName, Object> values = resolveValues(
        columns,
        expressions,
        schemaWithPseudoColumns,
        functionRegistry,
        config
    );

    if (dataSourceType == DataSourceType.KTABLE) {
      final String noValue = schemaWithPseudoColumns.key().stream()
          .map(Column::name)
          .filter(colName -> !values.containsKey(colName))
          .map(ColumnName::text)
          .collect(Collectors.joining(", "));

      if (!noValue.isEmpty()) {
        throw new KsqlException("Value for primary key column(s) "
            + noValue + " is required for tables");
      }
    }

    final long ts = (long) values.getOrDefault(SystemColumns.ROWTIME_NAME, clock.getAsLong());

    final GenericKey key = buildKey(schema, values);
    final GenericRow value = buildValue(schema, values);

    return KsqlGenericRecord.of(key, value, ts);
  }
@Test
  public void shouldBuildCoerceTypes() {
    // Given:
    final LogicalSchema schema = LogicalSchema.builder()
        .keyColumn(KEY, SqlTypes.BIGINT)
        .valueColumn(COL0, SqlTypes.BIGINT)
        .build();
    final List<ColumnName> names = ImmutableList.of(KEY, COL0);
    final Expression exp = new IntegerLiteral(1);

    // When:
    final KsqlGenericRecord record = recordFactory.build(
        names, ImmutableList.of(exp, exp), schema, DataSourceType.KSTREAM
    );

    // Then:
    assertThat(record, is(KsqlGenericRecord.of(
        GenericKey.genericKey(1L),
        GenericRow.genericRow(1L),
        0
    )));
  }