/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.table.runtime.batch;

//import org.apache.flink.api.common.typeinfo.Types;
//import org.apache.flink.api.java.DataSet;
//import org.apache.flink.api.java.ExecutionEnvironment;
//import org.apache.flink.formats.avro.generated.Address;
//import org.apache.flink.formats.avro.generated.Colors;
//import org.apache.flink.formats.avro.generated.Fixed16;
//import org.apache.flink.formats.avro.generated.Fixed2;
//import org.apache.flink.formats.avro.generated.User;
//import org.apache.flink.formats.avro.utils.AvroKryoSerializerUtils;
//import org.apache.flink.table.api.Table;
//import org.apache.flink.table.api.TableEnvironment;
//import org.apache.flink.table.api.java.BatchTableEnvironment;
//import org.apache.flink.table.runtime.utils.TableProgramsClusterTestBase;
//import org.apache.flink.test.util.TestBaseUtils;
//import org.apache.flink.types.Row;

//import org.apache.avro.util.Utf8;
//import org.joda.time.DateTime;
//import org.joda.time.LocalDate;
//import org.joda.time.LocalTime;
//import org.junit.Ignore;
//import org.junit.Test;
//import org.junit.runner.RunWith;
//import org.junit.runners.Parameterized;

//import java.math.BigDecimal;
//import java.nio.ByteBuffer;
//import java.util.ArrayList;
//import java.util.Arrays;
//import java.util.Collections;
//import java.util.HashMap;
//import java.util.List;

//import static org.junit.Assert.assertEquals;

/**
 * Tests for interoperability with Avro types.
 */
//@Ignore // TODO: should support this if we support transition from table to DataSet.
//@RunWith(Parameterized.class)
//public class AvroTypesITCase extends TableProgramsClusterTestBase {
//
//	private static final User USER_1 = User.newBuilder()
//		.setName("Charlie")
//		.setFavoriteColor("blue")
//		.setFavoriteNumber(null)
//		.setTypeBoolTest(false)
//		.setTypeDoubleTest(1.337d)
//		.setTypeNullTest(null)
//		.setTypeLongTest(1337L)
//		.setTypeArrayString(new ArrayList<>())
//		.setTypeArrayBoolean(new ArrayList<>())
//		.setTypeNullableArray(null)
//		.setTypeEnum(Colors.RED)
//		.setTypeMap(new HashMap<>())
//		.setTypeFixed(null)
//		.setTypeUnion(null)
//		.setTypeNested(
//			Address.newBuilder()
//				.setNum(42)
//				.setStreet("Bakerstreet")
//				.setCity("Berlin")
//				.setState("Berlin")
//				.setZip("12049").build())
//		.setTypeBytes(ByteBuffer.allocate(10))
//		.setTypeDate(LocalDate.parse("2014-03-01"))
//		.setTypeTimeMillis(LocalTime.parse("12:12:12"))
//		.setTypeTimeMicros(123456)
//		.setTypeTimestampMillis(DateTime.parse("2014-03-01T12:12:12.321Z"))
//		.setTypeTimestampMicros(123456L)
//		.setTypeDecimalBytes(ByteBuffer.wrap(BigDecimal.valueOf(2000, 2).unscaledValue().toByteArray()))
//		.setTypeDecimalFixed(new Fixed2(BigDecimal.valueOf(2000, 2).unscaledValue().toByteArray()))
//		.build();
//
//	private static final User USER_2 = User.newBuilder()
//		.setName("Whatever")
//		.setFavoriteNumber(null)
//		.setFavoriteColor("black")
//		.setTypeLongTest(42L)
//		.setTypeDoubleTest(0.0)
//		.setTypeNullTest(null)
//		.setTypeBoolTest(true)
//		.setTypeArrayString(Collections.singletonList("hello"))
//		.setTypeArrayBoolean(Collections.singletonList(true))
//		.setTypeEnum(Colors.GREEN)
//		.setTypeMap(new HashMap<>())
//		.setTypeFixed(new Fixed16())
//		.setTypeUnion(null)
//		.setTypeNested(null).setTypeDate(LocalDate.parse("2014-03-01"))
//		.setTypeBytes(ByteBuffer.allocate(10))
//		.setTypeTimeMillis(LocalTime.parse("12:12:12"))
//		.setTypeTimeMicros(123456)
//		.setTypeTimestampMillis(DateTime.parse("2014-03-01T12:12:12.321Z"))
//		.setTypeTimestampMicros(123456L)
//		.setTypeDecimalBytes(ByteBuffer.wrap(BigDecimal.valueOf(2000, 2).unscaledValue().toByteArray()))
//		.setTypeDecimalFixed(new Fixed2(BigDecimal.valueOf(2000, 2).unscaledValue().toByteArray()))
//		.build();
//
//	private static final User USER_3 = User.newBuilder()
//		.setName("Terminator")
//		.setFavoriteNumber(null)
//		.setFavoriteColor("yellow")
//		.setTypeLongTest(1L)
//		.setTypeDoubleTest(0.0)
//		.setTypeNullTest(null)
//		.setTypeBoolTest(false)
//		.setTypeArrayString(Collections.singletonList("world"))
//		.setTypeArrayBoolean(Collections.singletonList(false))
//		.setTypeEnum(Colors.GREEN)
//		.setTypeMap(new HashMap<>())
//		.setTypeFixed(new Fixed16())
//		.setTypeUnion(null)
//		.setTypeNested(null)
//		.setTypeBytes(ByteBuffer.allocate(10))
//		.setTypeDate(LocalDate.parse("2014-03-01"))
//		.setTypeTimeMillis(LocalTime.parse("12:12:12"))
//		.setTypeTimeMicros(123456)
//		.setTypeTimestampMillis(DateTime.parse("2014-03-01T12:12:12.321Z"))
//		.setTypeTimestampMicros(123456L)
//		.setTypeDecimalBytes(ByteBuffer.wrap(BigDecimal.valueOf(2000, 2).unscaledValue().toByteArray()))
//		.setTypeDecimalFixed(new Fixed2(BigDecimal.valueOf(2000, 2).unscaledValue().toByteArray()))
//		.build();
//
//	public AvroTypesITCase(
//		TestExecutionMode executionMode,
//		TableConfigMode tableConfigMode) {
//		super(executionMode, tableConfigMode);
//	}
//
//	@Test
//	public void testAvroToRow() throws Exception {
//		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
//		env.getConfig().registerTypeWithKryoSerializer(LocalDate.class, AvroKryoSerializerUtils.JodaLocalDateSerializer.class);
//		env.getConfig().registerTypeWithKryoSerializer(LocalTime.class, AvroKryoSerializerUtils.JodaLocalTimeSerializer.class);
//		BatchTableEnvironment tEnv = TableEnvironment.getTableEnvironment(env, config());
//
//		Table t = tEnv.fromDataSet(testData(env));
//		Table result = t.select("*");
//
//		List<Row> results = tEnv.toDataSet(result, Row.class).collect();
//		String expected =
//			"black,null,Whatever,[true],[hello],true,java.nio.HeapByteBuffer[pos=0 lim=10 cap=10]," +
//				"2014-03-01,java.nio.HeapByteBuffer[pos=0 lim=2 cap=2],[7, -48],0.0,GREEN," +
//				"[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],42,{},null,null,null,123456," +
//				"12:12:12.000,123456,2014-03-01T12:12:12.321Z,null\n" +
//				"blue,null,Charlie,[],[],false,java.nio.HeapByteBuffer[pos=0 lim=10 cap=10],2014-03-01," +
//				"java.nio.HeapByteBuffer[pos=0 lim=2 cap=2],[7, -48],1.337,RED,null,1337,{}," +
//				"{\"num\": 42, \"street\": \"Bakerstreet\", \"city\": \"Berlin\", \"state\": " +
//				"\"Berlin\", \"zip\": \"12049\"},null,null,123456,12:12:12.000,123456," +
//				"2014-03-01T12:12:12.321Z,null\n" +
//				"yellow,null,Terminator,[false],[world],false," +
//				"java.nio.HeapByteBuffer[pos=0 lim=10 cap=10],2014-03-01," +
//				"java.nio.HeapByteBuffer[pos=0 lim=2 cap=2],[7, -48],0.0,GREEN," +
//				"[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],1,{},null,null,null,123456," +
//				"12:12:12.000,123456,2014-03-01T12:12:12.321Z,null";
//		TestBaseUtils.compareResultAsText(results, expected);
//	}
//
//	@Test
//	public void testAvroStringAccess() throws Exception {
//		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
//		BatchTableEnvironment tEnv = TableEnvironment.getTableEnvironment(env, config());
//
//		Table t = tEnv.fromDataSet(testData(env));
//		Table result = t.select("name");
//		List<Utf8> results = tEnv.toDataSet(result, Types.GENERIC(Utf8.class)).collect();
//		String expected = "Charlie\n" +
//			"Terminator\n" +
//			"Whatever";
//		TestBaseUtils.compareResultAsText(results, expected);
//	}
//
//	@Test
//	public void testAvroObjectAccess() throws Exception {
//		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
//		BatchTableEnvironment tEnv = TableEnvironment.getTableEnvironment(env, config());
//
//		Table t = tEnv.fromDataSet(testData(env));
//		Table result = t
//			.filter("type_nested.isNotNull")
//			.select("type_nested.flatten()").as("city, num, state, street, zip");
//
//		List<Address> results = tEnv.toDataSet(result, Types.POJO(Address.class)).collect();
//		String expected = USER_1.getTypeNested().toString();
//		TestBaseUtils.compareResultAsText(results, expected);
//	}
//
//	@Test
//	public void testAvroToAvro() throws Exception {
//		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
//		BatchTableEnvironment tEnv = TableEnvironment.getTableEnvironment(env, config());
//
//		Table t = tEnv.fromDataSet(testData(env));
//		Table result = t.select("*");
//
//		List<User> results = tEnv.toDataSet(result, Types.POJO(User.class)).collect();
//		List<User> expected = Arrays.asList(USER_1, USER_2, USER_3);
//		assertEquals(expected, results);
//	}
//
//	private DataSet<User> testData(ExecutionEnvironment env) {
//		List<User> data = new ArrayList<>(3);
//		data.add(USER_1);
//		data.add(USER_2);
//		data.add(USER_3);
//		return env.fromCollection(data);
//	}
//}