MysqlDumpRowReader extends RowReader { @Override public Row nextRow() throws IOException { while (true) { int b = read(); switch (state) { case STATEMENT_START: if (b < 0) { return null; } else if (b == '-') { b = read(); if (b == '-') { state = State.SINGLE_LINE_COMMENT; } else { throw unexpectedToken('-', b); } } else if (b == '/') { b = read(); if (b == '*') { state = State.DELIMITED_COMMENT; } else { throw unexpectedToken('/', b); } } else if ((b >= 'A') && (b <= 'Z')) { addToField(b); state = State.STATEMENT_VERB; } else if ((b == ' ') || (b == '\r') || (b == '\n')) { } else { throw unexpectedToken(b); } break; case SINGLE_LINE_COMMENT: if (b < 0) { return null; } else if (b == '\n') { state = State.STATEMENT_START; } break; case DELIMITED_COMMENT: if (b < 0) { throw eofInTheMiddleOf("a comment"); } else if (b == '*') { b = read(); if (b == '/') { b = read(); if (b != ';') unread(b); state = State.STATEMENT_START; } else { unread(b); } } break; case STATEMENT_VERB: case INSERT: case INSERT_VALUES: if (b < 0) { throw eofInTheMiddleOf("a statement"); } else if ((b >= 'A') && (b <= 'Z')) { addToField(b); } else { if (b != ' ') unread(b); if (state == State.INSERT) { if (fieldMatches(into)) { clearField(); state = State.INSERT_TABLE; } else { throw new ExternalRowReaderException("Unrecognized statement INSERT " + decodeField()); } } else if (state == State.INSERT_VALUES) { if (fieldMatches(values)) { clearField(); state = State.NEXT_ROW_CTOR; } else { throw new ExternalRowReaderException("Unrecognized statement INSERT INTO " + decodeField()); } } else if (fieldMatches(lock) || fieldMatches(unlock)) { clearField(); state = State.IGNORED_STATEMENT; } else if (fieldMatches(insert)) { clearField(); state = State.INSERT; } else { throw new ExternalRowReaderException("Unrecognized statement " + decodeField()); } } break; case IGNORED_STATEMENT: if (b < 0) { throw eofInTheMiddleOf("a statement"); } else if (b == ';') { state = State.STATEMENT_START; } else if (b == '`') { state = State.IGNORED_BACKQUOTE; } break; case IGNORED_BACKQUOTE: if (b < 0) { throw eofInTheMiddleOf("a statement"); } else if (b == '`') { state = State.IGNORED_STATEMENT; } else if (b == '\\') { b = read(); } break; case INSERT_TABLE: if (b < 0) { throw eofInTheMiddleOf("a statement"); } else if (b == '`') { addToField(b); state = State.TABLE_BACKQUOTE; } else if ((b == '.') || ((b >= 'A') && (b <= 'Z')) || ((b >= 'a') && (b <= 'z')) || ((b >= '0') && (b <= '9')) || (b == '_')) { addToField(b); } else { if (b != ' ') unread(b); if (tableName == null) { tableName = copyField(); if (logger.isTraceEnabled()) { logger.trace("Original target table: {}", decodeField()); } } else if (!fieldMatches(tableName)) { throw new ExternalRowReaderException("INSERT INTO changed from " + decode(tableName) + " to " + decodeField() + ". Does file contain multiple tables?"); } clearField(); state = State.INSERT_VALUES; } break; case TABLE_BACKQUOTE: if (b < 0) { throw eofInTheMiddleOf("table name"); } else if (b == '`') { addToField(b); state = State.INSERT_TABLE; } else if (b == '\\') { addToField(b); b = read(); if (b >= 0) addToField(b); } else { addToField(b); } break; case NEXT_ROW_CTOR: if (b < 0) { throw eofInTheMiddleOf("a statement"); } else if (b == '(') { newRow(); state = State.NEXT_FIELD; } else { throw unexpectedToken(b); } break; case AFTER_ROW_CTOR: if (b < 0) { throw eofInTheMiddleOf("a statement"); } else if (b == ';') { state = State.STATEMENT_START; } else if (b == ',') { state = State.NEXT_ROW_CTOR; } else { throw unexpectedToken(b); } break; case NEXT_FIELD: if (b < 0) { throw eofInTheMiddleOf("a statement"); } else if (b == ')') { state = State.AFTER_ROW_CTOR; return finishRow(); } else if (b == '\'') { state = State.QUOTED_FIELD; } else if (b == ',') { addField(false); } else if ((b == ' ') || (b == '\r') || (b == '\n')) { } else { addToField(b); state = State.UNQUOTED_FIELD; } break; case UNQUOTED_FIELD: if (b < 0) { throw eofInTheMiddleOf("a statement"); } else if (b == ',') { addField(false); state = State.NEXT_FIELD; } else if (b == ')') { addField(false); state = State.AFTER_ROW_CTOR; return finishRow(); } else if (b == '\'') { throw new ExternalRowReaderException("Quote in the middle of a value"); } else { addToField(b); } break; case QUOTED_FIELD: if (b < 0) { throw eofInTheMiddleOf("quoted string"); } else if (b == '\'') { state = State.AFTER_QUOTED_FIELD; } else if (b == '\\') { b = read(); switch (b) { case -1: throw eofInTheMiddleOf("quoted string"); case 'n': b = '\n'; break; case 'r': b = '\r'; break; case 't': b = '\t'; break; } addToField(b); } else { addToField(b); } break; case AFTER_QUOTED_FIELD: if (b < 0) { throw eofInTheMiddleOf("a statement"); } else if (b == ',') { addField(true); state = State.NEXT_FIELD; } else if (b == ')') { addField(true); state = State.AFTER_ROW_CTOR; return finishRow(); } else { throw unexpectedToken(b); } break; } } } MysqlDumpRowReader(Table table, List<Column> columns,
                              InputStream inputStream, String encoding,
                              QueryContext queryContext, TypesTranslator typesTranslator); @Override Row nextRow();  }
@Test public void reader() throws Exception { SchemaFactory schemaFactory = new SchemaFactory("test"); AkibanInformationSchema ais = schemaFactory.aisWithRowDefs(DDL); Table t1 = ais.getTable("test", "t1"); InputStream istr = new FileInputStream(DUMP_FILE); MysqlDumpRowReader reader = new MysqlDumpRowReader(t1, t1.getColumns(), istr, "UTF-8", null, MTypesTranslator.INSTANCE); List<Row> rows = new ArrayList<>(); Row row; while ((row = reader.nextRow()) != null) rows.add(row); istr.close(); assertEquals("number of rows", ROWS.length, rows.size()); for (int i = 0; i < ROWS.length; i++) { Object[] orow = ROWS[i]; row = rows.get(i); assertEquals("row " + i + " size", orow.length, row.rowType().nFields()); for (int j = 0; j < orow.length; j++) { assertEquals("row " + i + " col " + j, orow[j], ValueSources.toObject(row.value(j))); } } }