[FLINK-2268] Remove Writable support from Scala TypeInformation Macro

flink-connectors/flink-hadoop-compatibility/src/test/scala/org/apache/flink/api/hadoopcompatibility/scala/ScalaWritableTypeInfoTest.scala

+/*
+ * 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.api.hadoopcompatibility.scala
+
+import org.apache.flink.api.java.typeutils.WritableTypeInfo
+import org.apache.flink.api.scala._
+import org.apache.flink.util.TestLogger
+import org.apache.hadoop.io.Text
+import org.junit.Assert._
+import org.junit.Test
+
+class ScalaWritableTypeInfoTest extends TestLogger {
+
+  @Test
+  def testWritableTypeInfo = {
+    val writableTypeInfo = createTypeInformation[Text]
+
+    assertTrue(writableTypeInfo.isInstanceOf[WritableTypeInfo[Text]])
+    assertEquals(classOf[Text], writableTypeInfo.getTypeClass)
+  }
+}

flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeInformationGen.scala

@@ -28,7 +28,6 @@ import org.apache.flink.api.java.typeutils._
 import org.apache.flink.api.java.typeutils.runtime.TupleSerializerBase
 import org.apache.flink.api.scala.typeutils._
 import org.apache.flink.types.Value
-import org.apache.hadoop.io.Writable
 
 import scala.collection.JavaConverters._
 import scala.collection.mutable
@@ -87,10 +86,6 @@ private[flink] trait TypeInformationGen[C <: Context] {
       mkValueTypeInfo(v)(c.WeakTypeTag(v.tpe).asInstanceOf[c.WeakTypeTag[Value]])
         .asInstanceOf[c.Expr[TypeInformation[T]]]
 
-    case d : WritableDescriptor =>
-      mkWritableTypeInfo(d)(c.WeakTypeTag(d.tpe).asInstanceOf[c.WeakTypeTag[Writable]])
-        .asInstanceOf[c.Expr[TypeInformation[T]]]
-
     case pojo: PojoDescriptor => mkPojo(pojo)
 
     case javaTuple: JavaTupleDescriptor => mkJavaTuple(javaTuple)
@@ -320,14 +315,6 @@ private[flink] trait TypeInformationGen[C <: Context] {
     }
   }
 
-  def mkWritableTypeInfo[T <: Writable : c.WeakTypeTag](
-      desc: UDTDescriptor): c.Expr[TypeInformation[T]] = {
-    val tpeClazz = c.Expr[Class[T]](Literal(Constant(desc.tpe)))
-    reify {
-      TypeExtractor.createHadoopWritableTypeInfo[T](tpeClazz.splice)
-    }
-  }
-
   def mkJavaTuple[T: c.WeakTypeTag](desc: JavaTupleDescriptor): c.Expr[TypeInformation[T]] = {
 
     val fieldsTrees = desc.fields map { f => mkTypeInfo(f)(c.WeakTypeTag(f.tpe)).tree }